133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""issue #150 本地账号认证单元测试（pytest / 纯标准库亦可 unittest 跑）。
|
||
|
||
覆盖：
|
||
- 密码哈希：hash/verify、恒定时间、盐随机（同密码两次哈希不同）
|
||
- UserStore：create/authenticate/角色校验/重复用户名/禁用账号/改密
|
||
- session：签发/校验/过期/篡改签名/伪造
|
||
- 守卫：require_auth 未登录 401、can_write readonly 403
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
import unittest
|
||
|
||
# 让 tests 能 import core.auth（仓库根在 ../../.. ）
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
||
if ROOT not in sys.path:
|
||
sys.path.insert(0, ROOT)
|
||
|
||
from core.auth import (UserStore, hash_password, verify_password,
|
||
issue_token, parse_token, require_auth, can_write, AuthError)
|
||
from core.auth.users import User
|
||
|
||
|
||
class TestPasswordHash(unittest.TestCase):
|
||
def test_hash_then_verify(self):
|
||
h = hash_password("S3cretPwd!")
|
||
self.assertTrue(h.startswith("pbkdf2_sha256$"))
|
||
self.assertTrue(verify_password("S3cretPwd!", h))
|
||
self.assertFalse(verify_password("wrong", h))
|
||
|
||
def test_salt_random(self):
|
||
# 同一密码两次哈希应不同（盐随机）
|
||
self.assertNotEqual(hash_password("S3cretPwd!"), hash_password("S3cretPwd!"))
|
||
|
||
def test_tampered_store_rejected(self):
|
||
h = hash_password("S3cretPwd!")
|
||
# 篡改 hash 段
|
||
scheme, it, salt, _ = h.split("$")
|
||
self.assertFalse(verify_password("S3cretPwd!", "%s$%s$%s$AAAA" % (scheme, it, salt)))
|
||
|
||
def test_empty_password(self):
|
||
with self.assertRaises(ValueError):
|
||
hash_password("")
|
||
|
||
|
||
class TestUserStore(unittest.TestCase):
|
||
def setUp(self):
|
||
self.store = UserStore()
|
||
self.user = self.store.create("alice", "password1", role="engineer")
|
||
|
||
def test_authenticate_success(self):
|
||
u = self.store.authenticate("alice", "password1")
|
||
self.assertIsNotNone(u)
|
||
self.assertEqual(u.id, self.user.id)
|
||
self.assertIsNotNone(u.last_login_at)
|
||
|
||
def test_authenticate_wrong_password(self):
|
||
self.assertIsNone(self.store.authenticate("alice", "nope"))
|
||
|
||
def test_authenticate_unknown_user(self):
|
||
self.assertIsNone(self.store.authenticate("bob", "password1"))
|
||
|
||
def test_duplicate_username(self):
|
||
with self.assertRaises(ValueError):
|
||
self.store.create("alice", "password2")
|
||
|
||
def test_short_password(self):
|
||
with self.assertRaises(ValueError):
|
||
self.store.create("carol", "123")
|
||
|
||
def test_invalid_role(self):
|
||
with self.assertRaises(ValueError):
|
||
self.store.create("dave", "password1", role="superuser")
|
||
|
||
def test_deactivate_blocks_login(self):
|
||
self.store.set_active(self.user.id, False)
|
||
self.assertIsNone(self.store.authenticate("alice", "password1"))
|
||
|
||
def test_set_password(self):
|
||
self.store.set_password(self.user.id, "brand-new-pwd")
|
||
self.assertIsNone(self.store.authenticate("alice", "password1"))
|
||
self.assertIsNotNone(self.store.authenticate("alice", "brand-new-pwd"))
|
||
|
||
def test_set_role(self):
|
||
self.store.set_role(self.user.id, "admin")
|
||
self.assertEqual(self.store.get(self.user.id).role, "admin")
|
||
|
||
|
||
class TestSession(unittest.TestCase):
|
||
def test_issue_and_parse(self):
|
||
tok = issue_token(42)
|
||
sess = parse_token(tok)
|
||
self.assertIsNotNone(sess)
|
||
self.assertEqual(sess.user_id, 42)
|
||
|
||
def test_expired(self):
|
||
tok = issue_token(1, ttl=-1) # 已过期
|
||
self.assertIsNone(parse_token(tok))
|
||
|
||
def test_tampered_sig(self):
|
||
tok = issue_token(1)
|
||
uid, exp, sig = tok.split(".")
|
||
bad = ".".join([uid, exp, "A" * len(sig)])
|
||
self.assertIsNone(parse_token(bad))
|
||
|
||
def test_garbage(self):
|
||
self.assertIsNone(parse_token("not.a.token"))
|
||
self.assertIsNone(parse_token(""))
|
||
|
||
|
||
class TestGuards(unittest.TestCase):
|
||
def setUp(self):
|
||
self.store = UserStore()
|
||
self.admin = self.store.create("admin", "password1", role="admin")
|
||
self.viewer = self.store.create("viewer", "password1", role="readonly")
|
||
|
||
def test_require_auth_no_token(self):
|
||
with self.assertRaises(AuthError):
|
||
require_auth({}, self.store)
|
||
|
||
def test_can_write_viewer_forbidden(self):
|
||
with self.assertRaises(AuthError):
|
||
can_write(self.viewer)
|
||
|
||
def test_can_write_admin_ok(self):
|
||
can_write(self.admin) # 不抛即通过
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|