223 lines
7.1 KiB
Python
223 lines
7.1 KiB
Python
"""Unit tests for webUI auth: argon2 hashing, refresh-token store, JWT pair."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import uuid
|
|
from collections.abc import Callable
|
|
|
|
import pytest
|
|
|
|
from contract_check.core.auth import (
|
|
RefreshTokenClaims,
|
|
TokenExpiredError,
|
|
TokenInvalidError,
|
|
create_refresh_token,
|
|
verify_refresh_token,
|
|
)
|
|
from contract_check.core.auth_refresh import RefreshTokenStore
|
|
from contract_check.core.auth_refresh_key import refresh_key
|
|
from contract_check.core.config import get_settings
|
|
from contract_check.core.security.passwords import (
|
|
PasswordError,
|
|
hash_password,
|
|
needs_rehash,
|
|
verify_password,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_settings_cache(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
get_settings.cache_clear()
|
|
monkeypatch.setenv("JWT_SECRET", "unit-test-secret-do-not-use-in-prod")
|
|
monkeypatch.setenv("JWT_ALGORITHM", "HS256")
|
|
monkeypatch.setenv("JWT_ACCESS_TTL_MINUTES", "1440")
|
|
monkeypatch.setenv("JWT_REFRESH_TTL_DAYS", "30")
|
|
get_settings.cache_clear()
|
|
yield
|
|
get_settings.cache_clear()
|
|
|
|
|
|
# ── argon2 ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_hash_and_verify_password_roundtrip() -> None:
|
|
h = hash_password("correct horse battery staple")
|
|
assert h != "correct horse battery staple"
|
|
assert h.startswith("$argon2id$")
|
|
assert verify_password("correct horse battery staple", h)
|
|
|
|
|
|
def test_verify_password_rejects_wrong_password() -> None:
|
|
h = hash_password("hunter2")
|
|
assert not verify_password("hunter3", h)
|
|
assert not verify_password("", h)
|
|
assert not verify_password("hunter2", "")
|
|
|
|
|
|
def test_hash_password_rejects_empty() -> None:
|
|
with pytest.raises(PasswordError):
|
|
hash_password("")
|
|
|
|
|
|
def test_verify_password_handles_malformed_hash() -> None:
|
|
assert not verify_password("anything", "not-a-real-hash")
|
|
assert not verify_password("anything", "$argon2id$truncated")
|
|
|
|
|
|
def test_needs_rehash_returns_false_for_fresh_hash() -> None:
|
|
h = hash_password("supersecret")
|
|
assert needs_rehash(h) is False
|
|
|
|
|
|
# ── refresh JWT ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_create_and_verify_refresh_token() -> None:
|
|
user_id = uuid.uuid4()
|
|
jti = uuid.uuid4().hex
|
|
token = create_refresh_token(user_id, jti)
|
|
claims = verify_refresh_token(token)
|
|
assert claims.sub == user_id
|
|
assert claims.jti == jti
|
|
assert claims.type == "refresh"
|
|
|
|
|
|
def test_verify_refresh_rejects_access_token() -> None:
|
|
from contract_check.core.auth import create_access_token
|
|
|
|
access = create_access_token(uuid.uuid4(), 42)
|
|
with pytest.raises(TokenInvalidError):
|
|
verify_refresh_token(access)
|
|
|
|
|
|
def test_verify_refresh_rejects_tampered_token() -> None:
|
|
token = create_refresh_token(uuid.uuid4(), uuid.uuid4().hex)
|
|
tampered = token[:-10] + ("A" if token[-10] != "A" else "B") + token[-9:]
|
|
with pytest.raises(TokenInvalidError):
|
|
verify_refresh_token(tampered)
|
|
|
|
|
|
def test_verify_refresh_rejects_expired(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("JWT_REFRESH_TTL_DAYS", "-1")
|
|
get_settings.cache_clear()
|
|
token = create_refresh_token(uuid.uuid4(), uuid.uuid4().hex)
|
|
with pytest.raises(TokenExpiredError):
|
|
verify_refresh_token(token)
|
|
|
|
|
|
def test_refresh_claims_roundtrip() -> None:
|
|
c = RefreshTokenClaims(sub=uuid.uuid4(), jti="abc", type="refresh", exp=123)
|
|
d = c.to_dict()
|
|
c2 = RefreshTokenClaims.from_dict(d)
|
|
assert c2.sub == c.sub
|
|
assert c2.jti == c.jti
|
|
assert c2.exp == c.exp
|
|
|
|
|
|
# ── refresh-token Redis store ─────────────────────────────────────────────────
|
|
|
|
|
|
class _FakeRedis:
|
|
"""In-memory async stand-in for redis.asyncio.Redis (SET/GET/DELETE/EXISTS/SCAN)."""
|
|
|
|
def __init__(self) -> None:
|
|
self._data: dict[str, str] = {}
|
|
self._ttls: dict[str, float] = {}
|
|
|
|
async def set(self, key: str, value: str, ex: int | None = None) -> None:
|
|
self._data[key] = value
|
|
if ex is not None:
|
|
self._ttls[key] = dt.datetime.now(tz=dt.UTC).timestamp() + ex
|
|
|
|
async def exists(self, key: str) -> int:
|
|
if key in self._data and self._unexpired(key):
|
|
return 1
|
|
return 0
|
|
|
|
async def delete(self, *keys: str) -> int:
|
|
removed = 0
|
|
for k in keys:
|
|
if k in self._data:
|
|
del self._data[k]
|
|
self._ttls.pop(k, None)
|
|
removed += 1
|
|
return removed
|
|
|
|
async def scan(
|
|
self, cursor: int = 0, match: str | None = None, count: int = 100
|
|
) -> tuple[int, list[str]]:
|
|
import fnmatch
|
|
|
|
all_keys = [k for k in self._data if self._unexpired(k)]
|
|
if match:
|
|
all_keys = [k for k in all_keys if fnmatch.fnmatch(k, match)]
|
|
return 0, all_keys
|
|
|
|
def _unexpired(self, key: str) -> bool:
|
|
if key not in self._ttls:
|
|
return True
|
|
return dt.datetime.now(tz=dt.UTC).timestamp() < self._ttls[key]
|
|
|
|
|
|
@pytest.fixture
|
|
def store_factory() -> Callable[[], tuple[RefreshTokenStore, _FakeRedis]]:
|
|
def _make() -> tuple[RefreshTokenStore, _FakeRedis]:
|
|
redis = _FakeRedis()
|
|
return RefreshTokenStore(redis, ttl_seconds=3600), redis
|
|
|
|
return _make
|
|
|
|
|
|
async def test_refresh_store_issue_and_validate(
|
|
store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
|
|
) -> None:
|
|
store, redis = store_factory()
|
|
user_id = uuid.uuid4()
|
|
|
|
jti = await store.issue(user_id)
|
|
|
|
assert isinstance(jti, str)
|
|
assert len(jti) == 32 # uuid4().hex
|
|
assert await store.is_valid(user_id, jti) is True
|
|
assert refresh_key(user_id, jti) in redis._data
|
|
|
|
|
|
async def test_refresh_store_revoke(
|
|
store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
|
|
) -> None:
|
|
store, _ = store_factory()
|
|
user_id = uuid.uuid4()
|
|
|
|
jti = await store.issue(user_id)
|
|
assert await store.revoke(user_id, jti) is True
|
|
assert await store.is_valid(user_id, jti) is False
|
|
# Idempotent revoke.
|
|
assert await store.revoke(user_id, jti) is False
|
|
|
|
|
|
async def test_refresh_store_revoke_all(
|
|
store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
|
|
) -> None:
|
|
store, _ = store_factory()
|
|
user_id = uuid.uuid4()
|
|
other = uuid.uuid4()
|
|
|
|
jti1 = await store.issue(user_id)
|
|
jti2 = await store.issue(user_id)
|
|
jti_other = await store.issue(other)
|
|
|
|
removed = await store.revoke_all(user_id)
|
|
|
|
assert removed == 2
|
|
assert await store.is_valid(user_id, jti1) is False
|
|
assert await store.is_valid(user_id, jti2) is False
|
|
# Other user unaffected.
|
|
assert await store.is_valid(other, jti_other) is True
|
|
|
|
|
|
async def test_refresh_store_is_valid_rejects_empty_jti(
|
|
store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
|
|
) -> None:
|
|
store, _ = store_factory()
|
|
assert await store.is_valid(uuid.uuid4(), "") is False
|