383 lines
16 KiB
Python
383 lines
16 KiB
Python
"""Unit tests for user, token, API-key, and provider identity boundaries."""
|
|
|
|
from contextlib import contextmanager
|
|
import datetime
|
|
import json
|
|
from unittest.mock import MagicMock
|
|
import uuid
|
|
|
|
import pytest
|
|
from psycopg2.errors import UniqueViolation
|
|
|
|
from core import api_keys as apiKeys
|
|
from core import auth, identity, users
|
|
|
|
|
|
USER_ID = "00000000-0000-0000-0000-000000000101"
|
|
KEY_ID = "00000000-0000-0000-0000-000000000201"
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, rows=None):
|
|
self.rows = list(rows or [])
|
|
self.executed = []
|
|
|
|
def execute(self, query, params=None):
|
|
self.executed.append((" ".join(query.split()), params))
|
|
|
|
def fetchone(self):
|
|
return self.rows.pop(0) if self.rows else None
|
|
|
|
|
|
def _cursorContext(cursor):
|
|
@contextmanager
|
|
def context():
|
|
yield cursor
|
|
|
|
return context
|
|
|
|
|
|
def test_user_timezone_and_password_validation(monkeypatch):
|
|
assert users.isValidTimezone("America/Chicago")
|
|
assert not users.isValidTimezone("not/a-zone")
|
|
assert users.normalizeTimezone(" UTC ") == "UTC"
|
|
with pytest.raises(ValueError, match="IANA"):
|
|
users.normalizeTimezone(None)
|
|
|
|
monkeypatch.setenv("DEFAULT_TIMEZONE", "America/Chicago")
|
|
assert users.getDefaultTimezone() == "America/Chicago"
|
|
assert users.validatePassword("long-enough")[0]
|
|
assert not users.validatePassword("short")[0]
|
|
assert not users.validatePassword("é" * 40)[0]
|
|
assert users.validatePassword(None) == (False, ["password"])
|
|
|
|
|
|
def test_user_lookup_helpers_and_registration_boundaries(monkeypatch):
|
|
select = MagicMock(
|
|
side_effect=[
|
|
{"id": USER_ID, "username": "alice", "timezone": "UTC"},
|
|
{"id": USER_ID},
|
|
None,
|
|
]
|
|
)
|
|
monkeypatch.setattr(users.postgres, "select_one", select)
|
|
monkeypatch.setattr(users.postgres, "exists", MagicMock(side_effect=[False, True]))
|
|
assert users.getUser(USER_ID)["username"] == "alice"
|
|
assert users.getUserUUID("alice") == USER_ID
|
|
assert users.getUserUUID("missing") is False
|
|
assert users.isUsernameAvailable("alice")
|
|
assert users.doesUserUUIDExist(USER_ID)
|
|
|
|
monkeypatch.setattr(users, "isUsernameAvailable", lambda _name: True)
|
|
monkeypatch.setattr(users.bcrypt, "gensalt", lambda: b"salt")
|
|
monkeypatch.setattr(users.bcrypt, "hashpw", lambda value, _salt: b"hash:" + value)
|
|
create = MagicMock()
|
|
monkeypatch.setattr(users, "createUser", create)
|
|
assert users.registerUser(
|
|
" alice ",
|
|
"password123",
|
|
{"timezone": "UTC", "id": "attacker", "admin": True},
|
|
)
|
|
created = create.call_args.args[0]
|
|
assert created["username"] == "alice"
|
|
assert created["password_hashed"] == b"hash:password123"
|
|
assert created["timezone"] == "UTC"
|
|
assert created["id"] != "attacker" and "admin" not in created
|
|
|
|
monkeypatch.setattr(users, "isUsernameAvailable", lambda _name: False)
|
|
assert users.registerUser("alice", "password123") is False
|
|
monkeypatch.setattr(users, "isUsernameAvailable", lambda _name: True)
|
|
create.side_effect = UniqueViolation()
|
|
assert users.registerUser("alice", "password123") is False
|
|
|
|
|
|
@pytest.mark.parametrize("username", [None, " ", "x" * 256])
|
|
def test_registration_rejects_invalid_usernames(username):
|
|
with pytest.raises(ValueError, match="username"):
|
|
users.registerUser(username, "password123")
|
|
|
|
|
|
def test_user_creation_updates_password_and_delete(monkeypatch):
|
|
monkeypatch.setattr(users, "getUser", lambda _user: {"id": USER_ID})
|
|
update = MagicMock(return_value=[{"id": USER_ID}])
|
|
delete = MagicMock(return_value=[{"id": USER_ID}])
|
|
insert = MagicMock(return_value={"id": USER_ID})
|
|
monkeypatch.setattr(users.postgres, "update", update)
|
|
monkeypatch.setattr(users.postgres, "delete", delete)
|
|
monkeypatch.setattr(users.postgres, "insert", insert)
|
|
|
|
assert users.updateUser(USER_ID, {"timezone": "UTC", "username": "ignored"})
|
|
update.assert_called_with("users", {"timezone": "UTC"}, {"id": USER_ID})
|
|
assert not users.updateUser(USER_ID, {"username": "ignored"})
|
|
|
|
monkeypatch.setattr(users.bcrypt, "gensalt", lambda: b"salt")
|
|
monkeypatch.setattr(users.bcrypt, "hashpw", lambda _value, _salt: b"new-hash")
|
|
assert users.changePassword(USER_ID, "new-password")
|
|
assert users.deleteUser(USER_ID)
|
|
|
|
valid = {
|
|
"id": USER_ID,
|
|
"username": "alice",
|
|
"password_hashed": b"hash",
|
|
"timezone": "UTC",
|
|
"unexpected": True,
|
|
}
|
|
assert users.createUser(valid) == {"id": USER_ID}
|
|
assert "unexpected" not in insert.call_args.args[1]
|
|
assert users.validateUser(valid)[0]
|
|
assert not users.validateUser({"id": "bad", "timezone": "bad"})[0]
|
|
|
|
|
|
def test_login_token_creation_decoding_and_protected_claims(monkeypatch):
|
|
monkeypatch.setenv("JWT_SECRET", "test-secret")
|
|
monkeypatch.setattr(users, "doesUserUUIDExist", lambda value: value == USER_ID)
|
|
monkeypatch.setattr(users, "getUserFirstName", lambda _value: "Alice")
|
|
encode = MagicMock(return_value="encoded-token")
|
|
monkeypatch.setattr(auth.jwt, "encode", encode)
|
|
|
|
assert auth.createLoginToken(
|
|
USER_ID,
|
|
expiresIn=60,
|
|
extraClaims={"provider": "discord", "sub": "attacker"},
|
|
) == "encoded-token"
|
|
payload = encode.call_args.args[0]
|
|
assert payload["sub"] == USER_ID and payload["provider"] == "discord"
|
|
assert payload["exp"] > payload["iat"]
|
|
with pytest.raises(ValueError, match="positive"):
|
|
auth.createLoginToken(USER_ID, expiresIn=0)
|
|
with pytest.raises(ValueError, match="extra claims"):
|
|
auth.createLoginToken(USER_ID, extraClaims="bad")
|
|
|
|
monkeypatch.setattr(auth.jwt, "decode", lambda *_args, **_kwargs: {"sub": USER_ID})
|
|
assert auth.decodeLoginToken("token")["sub"] == USER_ID
|
|
monkeypatch.setattr(auth.jwt, "decode", lambda *_args, **_kwargs: {"sub": 123})
|
|
assert auth.decodeLoginToken("token") is None
|
|
monkeypatch.delenv("JWT_SECRET")
|
|
assert auth.decodeLoginToken("token") is None
|
|
with pytest.raises(RuntimeError, match="JWT_SECRET"):
|
|
auth.createLoginToken(USER_ID)
|
|
|
|
|
|
def test_password_login_and_account_removal(monkeypatch):
|
|
monkeypatch.setattr(users, "getUserUUID", lambda name: USER_ID if name == "alice" else False)
|
|
monkeypatch.setattr(auth, "getUserpasswordHash", lambda _value: b"stored")
|
|
monkeypatch.setattr(auth.bcrypt, "checkpw", lambda password, hashed: password == b"password123")
|
|
monkeypatch.setattr(auth, "createLoginToken", lambda _value: "login-token")
|
|
assert auth.getLoginToken("alice", "password123") == "login-token"
|
|
assert auth.getLoginToken("alice", "wrong-password") is False
|
|
assert auth.getLoginToken(None, "password123") is False
|
|
|
|
monkeypatch.setattr(users, "deleteUser", MagicMock(return_value=True))
|
|
assert auth.unregisterUser(USER_ID, "password123")
|
|
assert not auth.unregisterUser(USER_ID, "wrong-password")
|
|
|
|
|
|
def test_bearer_authentication_and_principal_helpers(monkeypatch):
|
|
assert auth.getBearerToken("Bearer token") == "token"
|
|
assert auth.getBearerToken("basic token") is None
|
|
assert auth.getBearerToken(None) is None
|
|
|
|
monkeypatch.setattr(auth, "decodeLoginToken", lambda _token: {"sub": USER_ID})
|
|
principal = auth.authenticateBearerToken("Bearer a.b.c")
|
|
assert principal["authentication"] == "jwt"
|
|
assert auth.isUserPrincipal(principal, USER_ID, requireLogin=True)
|
|
assert auth.authenticateBearerToken(
|
|
"Bearer a.b.c", requiredScopes=["service:scope"]
|
|
) is None
|
|
|
|
service = {
|
|
"type": "service",
|
|
"scopes": ["jobs:claim"],
|
|
"authentication": "api_key",
|
|
}
|
|
monkeypatch.setattr(apiKeys, "authenticateApiKey", lambda *_args, **_kwargs: service)
|
|
assert auth.authenticateBearerToken("Bearer service-key") == service
|
|
assert auth.authenticateBearerToken("Bearer service-key", allowService=False) is None
|
|
assert auth.hasServiceScope(service, "jobs:claim")
|
|
assert not auth.hasServiceScope(principal, "jobs:claim")
|
|
|
|
|
|
def test_api_key_normalization_public_shape_and_ids(monkeypatch):
|
|
assert apiKeys._normalizeScopes("one, two one") == ["one", "two"]
|
|
assert apiKeys._normalizeScopes(None) == []
|
|
with pytest.raises(ValueError, match="scopes"):
|
|
apiKeys._normalizeScopes(123)
|
|
with pytest.raises(ValueError, match="future"):
|
|
apiKeys._normalizeExpiry("2000-01-01T00:00:00Z")
|
|
|
|
record = {"id": KEY_ID, "key_hash": "secret", "scopes": '["one"]'}
|
|
assert apiKeys._publicKey(record, includeSecret="raw") == {
|
|
"id": KEY_ID,
|
|
"scopes": ["one"],
|
|
"key": "raw",
|
|
}
|
|
assert apiKeys._keyID(KEY_ID) == KEY_ID
|
|
assert apiKeys._keyID("bad") is None
|
|
monkeypatch.setattr(apiKeys.postgres, "select_one", MagicMock(return_value=record))
|
|
assert apiKeys.getApiKey("bad") is None
|
|
assert apiKeys.getApiKey(KEY_ID)["id"] == KEY_ID
|
|
|
|
|
|
def test_api_key_creation_and_owner_rules(monkeypatch):
|
|
monkeypatch.setattr(users, "doesUserUUIDExist", lambda value: value == USER_ID)
|
|
secret = "x" * 32
|
|
stored = {
|
|
"id": KEY_ID,
|
|
"name": "cli",
|
|
"key_type": "user",
|
|
"user_uuid": USER_ID,
|
|
"key_hash": apiKeys._hashToken(secret),
|
|
"scopes": [],
|
|
}
|
|
cursor = FakeCursor([stored])
|
|
monkeypatch.setattr(apiKeys.postgres, "get_cursor", _cursorContext(cursor))
|
|
created = apiKeys.createApiKey(
|
|
"cli", "user", userUUID=USER_ID, secret=secret
|
|
)
|
|
assert created["key"] == secret and "key_hash" not in created
|
|
assert cursor.executed[0][1]["key_prefix"] == secret[:20]
|
|
|
|
with pytest.raises(ValueError, match="user does not exist"):
|
|
apiKeys.createUserApiKey(str(uuid.uuid4()), "missing")
|
|
with pytest.raises(ValueError, match="cannot have a service"):
|
|
apiKeys.createApiKey(
|
|
"bad", "user", userUUID=USER_ID, serviceName="service"
|
|
)
|
|
with pytest.raises(ValueError, match="service name"):
|
|
apiKeys.createApiKey("bad", "service", scopes=[])
|
|
with pytest.raises(ValueError, match="cannot have a user"):
|
|
apiKeys.createApiKey(
|
|
"bad", "service", userUUID=USER_ID, serviceName="service"
|
|
)
|
|
|
|
|
|
def test_api_key_authentication_scope_expiry_and_revoke(monkeypatch):
|
|
secret = "service-secret-that-is-at-least-32-characters"
|
|
record = {
|
|
"id": KEY_ID,
|
|
"key_type": "service",
|
|
"user_uuid": None,
|
|
"service_name": "worker",
|
|
"key_hash": apiKeys._hashToken(secret),
|
|
"scopes": ["jobs:claim"],
|
|
"expires_at": None,
|
|
}
|
|
select = MagicMock(return_value=[record])
|
|
update = MagicMock()
|
|
monkeypatch.setattr(apiKeys.postgres, "select", select)
|
|
monkeypatch.setattr(apiKeys.postgres, "update", update)
|
|
principal = apiKeys.authenticateApiKey(secret, ["jobs:claim"])
|
|
assert principal["service_name"] == "worker"
|
|
assert update.called
|
|
assert apiKeys.authenticateApiKey("short") is None
|
|
assert apiKeys.authenticateApiKey(secret, ["jobs:deliver"]) is None
|
|
|
|
expired = dict(record, expires_at=apiKeys._utcNow() - datetime.timedelta(seconds=1))
|
|
select.return_value = [expired]
|
|
assert apiKeys.authenticateApiKey(secret) is None
|
|
select.return_value = []
|
|
assert apiKeys.authenticateApiKey(secret) is None
|
|
|
|
execute = MagicMock(return_value=[{"id": KEY_ID}])
|
|
monkeypatch.setattr(apiKeys.postgres, "execute", execute)
|
|
assert apiKeys.revokeApiKey(KEY_ID, serviceName="worker")
|
|
assert not apiKeys.revokeApiKey("invalid")
|
|
|
|
|
|
def test_service_key_bootstrap_is_idempotent_configuration(monkeypatch):
|
|
monkeypatch.delenv("BOT_API_KEY", raising=False)
|
|
assert apiKeys.bootstrapServiceApiKey() is None
|
|
secret = "configured-service-key-at-least-32-characters"
|
|
monkeypatch.setenv("BOT_API_KEY", secret)
|
|
monkeypatch.setenv("BOT_API_KEY_SCOPES", "discord:session, outbox:claim")
|
|
record = {
|
|
"id": KEY_ID,
|
|
"key_hash": apiKeys._hashToken(secret),
|
|
"scopes": json.dumps(["discord:session", "outbox:claim"]),
|
|
}
|
|
cursor = FakeCursor([record])
|
|
monkeypatch.setattr(apiKeys.postgres, "get_cursor", _cursorContext(cursor))
|
|
result = apiKeys.bootstrapServiceApiKey()
|
|
assert result["scopes"] == ["discord:session", "outbox:claim"]
|
|
assert len(cursor.executed) == 2
|
|
|
|
|
|
def test_identity_normalization_enrollment_and_lookup(monkeypatch):
|
|
assert identity._normalizeProvider(" Discord ") == "discord"
|
|
assert identity._normalizeProviderUserID(123) == "123"
|
|
assert identity._normalizeDisplayName(" Alice ") == "Alice"
|
|
with pytest.raises(ValueError, match="provider"):
|
|
identity._normalizeProvider("bad provider")
|
|
with pytest.raises(ValueError, match="provider user ID"):
|
|
identity._normalizeProviderUserID("")
|
|
|
|
monkeypatch.setenv("DISCORD_ENROLLMENT_MODE", "allowlist")
|
|
monkeypatch.setenv("DISCORD_ALLOWLIST", "123, 456\n789")
|
|
assert identity.getDiscordAllowlist() == {"123", "456", "789"}
|
|
assert identity.isDiscordEnrollmentAllowed("456")
|
|
assert not identity.isDiscordEnrollmentAllowed("000")
|
|
monkeypatch.setenv("DISCORD_ENROLLMENT_MODE", "open")
|
|
assert identity.isDiscordEnrollmentAllowed("000")
|
|
|
|
monkeypatch.setattr(
|
|
identity.postgres,
|
|
"select_one",
|
|
MagicMock(return_value={"user_uuid": USER_ID}),
|
|
)
|
|
monkeypatch.setattr(users, "getUser", MagicMock(return_value={"id": USER_ID}))
|
|
assert identity.getDiscordUser("123") == {"id": USER_ID}
|
|
|
|
|
|
def test_link_provider_identity_existing_and_new_paths(monkeypatch):
|
|
monkeypatch.setattr(users, "doesUserUUIDExist", lambda _value: True)
|
|
existing = {
|
|
"id": "identity-one",
|
|
"user_uuid": USER_ID,
|
|
"display_name": "Old",
|
|
}
|
|
monkeypatch.setattr(identity, "getProviderIdentity", MagicMock(return_value=existing))
|
|
update = MagicMock(return_value=[dict(existing, display_name="New")])
|
|
monkeypatch.setattr(identity.postgres, "update", update)
|
|
assert identity.linkProviderIdentity(
|
|
USER_ID, "discord", "123", displayName="New"
|
|
)["display_name"] == "New"
|
|
|
|
identity.getProviderIdentity.return_value = dict(existing, user_uuid=str(uuid.uuid4()))
|
|
with pytest.raises(ValueError, match="another user"):
|
|
identity.linkProviderIdentity(USER_ID, "discord", "123")
|
|
|
|
identity.getProviderIdentity.return_value = None
|
|
inserted = {"id": "identity-two", "user_uuid": USER_ID}
|
|
monkeypatch.setattr(identity.postgres, "insert", MagicMock(return_value=inserted))
|
|
assert identity.linkDiscordIdentity(USER_ID, "456") == inserted
|
|
monkeypatch.setattr(identity.postgres, "delete", MagicMock(return_value=[inserted]))
|
|
assert identity.unlinkProviderIdentity(USER_ID, "discord", "456")
|
|
|
|
|
|
def test_get_or_create_discord_user_existing_denied_and_created(monkeypatch):
|
|
monkeypatch.setattr(users, "getDefaultTimezone", lambda: "UTC")
|
|
existingIdentity = {
|
|
"id": "identity-one",
|
|
"user_uuid": USER_ID,
|
|
"display_name": "Old",
|
|
}
|
|
monkeypatch.setattr(
|
|
identity, "getProviderIdentity", MagicMock(return_value=existingIdentity)
|
|
)
|
|
monkeypatch.setattr(users, "getUser", MagicMock(return_value={"id": USER_ID}))
|
|
update = MagicMock()
|
|
monkeypatch.setattr(identity.postgres, "update", update)
|
|
assert identity.getOrCreateDiscordUser("123", "New") == {"id": USER_ID}
|
|
assert update.called
|
|
|
|
identity.getProviderIdentity.return_value = None
|
|
monkeypatch.setattr(identity, "isDiscordEnrollmentAllowed", lambda _value: False)
|
|
assert identity.getOrCreateDiscordUser("999") is None
|
|
|
|
monkeypatch.setattr(identity, "isDiscordEnrollmentAllowed", lambda _value: True)
|
|
cursor = FakeCursor([None, {"id": USER_ID, "timezone": "UTC"}])
|
|
monkeypatch.setattr(identity.postgres, "get_cursor", _cursorContext(cursor))
|
|
created = identity.getOrCreateDiscordUser("999", "New")
|
|
assert created["id"] == USER_ID
|
|
assert len(cursor.executed) == 4
|