514 lines
17 KiB
Python
514 lines
17 KiB
Python
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pyotp
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
|
|
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
|
|
|
|
|
|
def issue_challenge(
|
|
store: SQLiteIdentityStore,
|
|
*,
|
|
principal_id: str,
|
|
challenge_id: str,
|
|
domain: str,
|
|
now: int,
|
|
) -> None:
|
|
store.issue_owner_challenge(
|
|
challenge_id=challenge_id,
|
|
owner_principal_id=principal_id,
|
|
proposal_id="proposal-one",
|
|
proposal_revision=1,
|
|
payload_digest="a" * 64,
|
|
diff_digest="b" * 64,
|
|
domains=(domain,),
|
|
expires_at=now + 5 * 60,
|
|
now=now,
|
|
)
|
|
|
|
|
|
def test_credentials_are_hashed_revocable_and_project_scoped(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "private" / "identity.sqlite3")
|
|
token = store.provision(
|
|
credential_id="agent-key-1",
|
|
principal_id="agent-pilot",
|
|
roles=("agent",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
|
|
principal = store.authenticate(token)
|
|
assert principal is not None
|
|
assert principal.principal_id == "agent-pilot"
|
|
assert principal.credential_id == "agent-key-1"
|
|
assert principal.can_read_project("dome-performance")
|
|
assert not principal.can_read_project("interactive-video")
|
|
assert token.encode() not in (tmp_path / "private" / "identity.sqlite3").read_bytes()
|
|
assert store.revoke("agent-key-1")
|
|
assert store.authenticate(token) is None
|
|
assert not store.revoke("agent-key-1")
|
|
|
|
|
|
def test_admin_role_can_read_without_project_grant(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
token = store.provision(
|
|
credential_id="admin-key",
|
|
principal_id="operator-1",
|
|
roles=("admin",),
|
|
)
|
|
principal = store.authenticate(token)
|
|
assert principal is not None
|
|
assert principal.can_read_project("dome-performance")
|
|
|
|
|
|
def test_publication_authority_snapshot_detects_grant_or_role_change(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
token = store.provision(
|
|
credential_id="agent-key",
|
|
principal_id="agent-one",
|
|
roles=("agent",),
|
|
project_ids=("dome-performance", "shared"),
|
|
)
|
|
principal = store.authenticate(token)
|
|
assert principal is not None
|
|
assert store.is_current(
|
|
principal.principal_id,
|
|
principal.credential_id,
|
|
roles=principal.roles,
|
|
project_ids=principal.project_ids,
|
|
)
|
|
assert not store.is_current(
|
|
principal.principal_id,
|
|
principal.credential_id,
|
|
roles=principal.roles,
|
|
project_ids=frozenset({"dome-performance"}),
|
|
)
|
|
|
|
|
|
def test_rotation_revokes_old_key_and_preserves_attribution_and_grants(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
old_token = store.provision(
|
|
credential_id="agent-key-old",
|
|
principal_id="agent-pilot",
|
|
roles=("agent",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
new_token = store.rotate("agent-key-old", "agent-key-new")
|
|
assert store.authenticate(old_token) is None
|
|
principal = store.authenticate(new_token)
|
|
assert principal is not None
|
|
assert principal.principal_id == "agent-pilot"
|
|
assert principal.credential_id == "agent-key-new"
|
|
assert principal.can_read_project("dome-performance")
|
|
|
|
|
|
def test_browser_sessions_are_human_only_csrf_protected_and_revocable(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
viewer_key = store.provision(
|
|
credential_id="viewer-key",
|
|
principal_id="viewer-one",
|
|
roles=("viewer",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
agent_key = store.provision(
|
|
credential_id="agent-key",
|
|
principal_id="agent-one",
|
|
roles=("agent",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
with pytest.raises(PermissionError, match="human browser identity"):
|
|
store.create_browser_session(agent_key)
|
|
|
|
session = store.create_browser_session(viewer_key)
|
|
principal = store.authenticate_browser_session(session.session_token)
|
|
assert principal is not None
|
|
assert principal.auth_method == "browser_session"
|
|
assert principal.can_read_project("dome-performance")
|
|
assert not principal.can_submit_project("dome-performance")
|
|
assert store.browser_csrf_valid(session.session_token, session.csrf_token)
|
|
assert not store.browser_csrf_valid(session.session_token, "gwc_invalid")
|
|
assert store.revoke_browser_session(session.session_token)
|
|
assert store.authenticate_browser_session(session.session_token) is None
|
|
assert not store.browser_csrf_valid(session.session_token, session.csrf_token)
|
|
|
|
admin_key = store.provision(
|
|
credential_id="human-admin-key",
|
|
principal_id="human-admin",
|
|
roles=("admin",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
admin_session = store.create_browser_session(admin_key)
|
|
browser_admin = store.authenticate_browser_session(admin_session.session_token)
|
|
assert browser_admin is not None and not browser_admin.can_submit_project("dome-performance")
|
|
|
|
|
|
def test_revoking_browser_credential_invalidates_its_sessions(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
token = store.provision(
|
|
credential_id="human-key",
|
|
principal_id="human-one",
|
|
roles=("editor",),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
session = store.create_browser_session(token)
|
|
assert store.authenticate_browser_session(session.session_token) is not None
|
|
assert store.revoke("human-key")
|
|
assert store.authenticate_browser_session(session.session_token) is None
|
|
|
|
|
|
def test_identity_store_rejects_symlinked_database_path(tmp_path: Path) -> None:
|
|
target = tmp_path / "real"
|
|
target.mkdir()
|
|
link = tmp_path / "linked"
|
|
link.symlink_to(target, target_is_directory=True)
|
|
with pytest.raises(ValueError, match="symbolic links"):
|
|
SQLiteIdentityStore(link / "identity.sqlite3")
|
|
|
|
|
|
def test_owner_session_is_challenge_bound_short_lived_and_separate_from_agent_auth(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = SQLiteIdentityStore(
|
|
tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
|
|
)
|
|
secret = pyotp.random_base32()
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=secret,
|
|
project_ids=("dome-performance",),
|
|
)
|
|
database_bytes = (tmp_path / "identity.sqlite3").read_bytes()
|
|
assert secret.encode() not in database_bytes
|
|
assert b"correct horse battery staple" not in database_bytes
|
|
now = int(time.time())
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-1",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
session = store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_code=pyotp.TOTP(secret).at(now),
|
|
challenge_id="challenge-1",
|
|
audience="https://gewu.example",
|
|
now=now,
|
|
)
|
|
assert session is not None
|
|
assert session.expires_at.timestamp() == now + 5 * 60
|
|
principal = store.authenticate_owner_session(
|
|
session.session_token,
|
|
challenge_id="challenge-1",
|
|
audience="https://gewu.example",
|
|
)
|
|
assert principal is not None
|
|
assert principal.roles == frozenset({"owner"})
|
|
assert principal.auth_method == "owner_session"
|
|
assert principal.can_read_project("dome-performance")
|
|
assert not principal.can_submit_project("dome-performance")
|
|
assert store.authenticate(session.session_token) is None
|
|
assert (
|
|
store.authenticate_owner_session(
|
|
session.session_token,
|
|
challenge_id="challenge-2",
|
|
audience="https://gewu.example",
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
store.authenticate_owner_session(
|
|
session.session_token,
|
|
challenge_id="challenge-1",
|
|
audience="https://other.example",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_owner_totp_cannot_be_replayed_and_login_failures_are_rate_limited(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = SQLiteIdentityStore(
|
|
tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
|
|
)
|
|
secret = pyotp.random_base32()
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=secret,
|
|
project_ids=("dome-performance",),
|
|
)
|
|
now = int(time.time())
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-1",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
code = pyotp.TOTP(secret).at(now)
|
|
args = {
|
|
"principal_id": "project-owner",
|
|
"password": "correct horse battery staple",
|
|
"totp_code": code,
|
|
"challenge_id": "challenge-1",
|
|
"audience": "https://gewu.example",
|
|
"now": now,
|
|
}
|
|
assert store.authenticate_owner(**args) is not None
|
|
assert store.authenticate_owner(**args) is None
|
|
|
|
for attempt in range(5):
|
|
challenge_id = f"challenge-bad-{attempt}"
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id=challenge_id,
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
assert (
|
|
store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="incorrect password",
|
|
totp_code="000000",
|
|
challenge_id=challenge_id,
|
|
audience="https://gewu.example",
|
|
now=now,
|
|
)
|
|
is None
|
|
)
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-final",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
assert (
|
|
store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="incorrect password",
|
|
totp_code="000000",
|
|
challenge_id="challenge-final",
|
|
audience="https://gewu.example",
|
|
now=now,
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_owner_operations_stay_disabled_without_encryption_key(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
with pytest.raises(RuntimeError, match="encryption key is not configured"):
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=pyotp.random_base32(),
|
|
project_ids=(),
|
|
)
|
|
assert (
|
|
store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_code="000000",
|
|
challenge_id="challenge-1",
|
|
audience="https://gewu.example",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_owner_challenge_retry_is_bound_to_payload_diff_and_granted_domains(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = SQLiteIdentityStore(
|
|
tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
|
|
)
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=pyotp.random_base32(),
|
|
project_ids=("dome-performance",),
|
|
)
|
|
now = int(time.time())
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-1",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-1",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
with pytest.raises(ValueError, match="different content"):
|
|
store.issue_owner_challenge(
|
|
challenge_id="challenge-1",
|
|
owner_principal_id="project-owner",
|
|
proposal_id="proposal-one",
|
|
proposal_revision=1,
|
|
payload_digest="c" * 64,
|
|
diff_digest="b" * 64,
|
|
domains=("dome-performance",),
|
|
expires_at=now + 300,
|
|
now=now,
|
|
)
|
|
|
|
|
|
def test_owner_decision_is_exact_diff_bound_and_idempotent(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(
|
|
tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
|
|
)
|
|
secret = pyotp.random_base32()
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=secret,
|
|
project_ids=("dome-performance",),
|
|
)
|
|
now = int(time.time())
|
|
diff_json = json.dumps({"files": [{"path": "projects/dome/nodes/a.yaml"}]})
|
|
diff_digest = hashlib.sha256(diff_json.encode()).hexdigest()
|
|
store.issue_owner_challenge(
|
|
challenge_id="challenge-bound",
|
|
owner_principal_id="project-owner",
|
|
proposal_id="proposal-one",
|
|
proposal_revision=1,
|
|
payload_digest="a" * 64,
|
|
diff_digest=diff_digest,
|
|
domains=("dome-performance",),
|
|
expires_at=now + 300,
|
|
diff_json=diff_json,
|
|
now=now,
|
|
)
|
|
token = store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_code=pyotp.TOTP(secret).at(now),
|
|
challenge_id="challenge-bound",
|
|
audience="gewu-test",
|
|
now=now,
|
|
)
|
|
assert token is not None
|
|
assert (
|
|
store.authenticate_owner_session(
|
|
token.session_token, challenge_id="challenge-bound", audience="gewu-test"
|
|
)
|
|
is not None
|
|
)
|
|
challenge = store.get_owner_challenge("challenge-bound", principal_id="project-owner")
|
|
assert challenge is not None
|
|
assert challenge["diff_json"] == diff_json
|
|
|
|
arguments = {
|
|
"challenge_id": "challenge-bound",
|
|
"principal_id": "project-owner",
|
|
"proposal_revision": 1,
|
|
"payload_digest": "a" * 64,
|
|
"diff_digest": diff_digest,
|
|
"decision": "authorize",
|
|
"rationale": "Reviewed exact proposed files and approve the policy exception.",
|
|
"owner_session_id": hashlib.sha256(token.session_token.encode()).hexdigest(),
|
|
"audience": "gewu-test",
|
|
"now": now,
|
|
}
|
|
with pytest.raises(ValueError, match="does not match"):
|
|
store.decide_owner_challenge(**{**arguments, "diff_digest": "b" * 64})
|
|
with pytest.raises(PermissionError, match="authority is unavailable"):
|
|
store.decide_owner_challenge(**{**arguments, "owner_session_id": "0" * 64})
|
|
first = store.decide_owner_challenge(**arguments)
|
|
assert first["status"] == "authorized"
|
|
assert store.decide_owner_challenge(**arguments) == first
|
|
with pytest.raises(ValueError, match="different decision"):
|
|
store.decide_owner_challenge(**{**arguments, "decision": "reject"})
|
|
with pytest.raises(PermissionError, match="every affected domain"):
|
|
store.issue_owner_challenge(
|
|
challenge_id="challenge-out-of-scope",
|
|
owner_principal_id="project-owner",
|
|
proposal_id="proposal-one",
|
|
proposal_revision=1,
|
|
payload_digest="a" * 64,
|
|
diff_digest="b" * 64,
|
|
domains=("interactive-video",),
|
|
expires_at=now + 300,
|
|
now=now,
|
|
)
|
|
|
|
|
|
def test_owner_grant_change_and_revocation_invalidate_sessions(tmp_path: Path) -> None:
|
|
store = SQLiteIdentityStore(
|
|
tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
|
|
)
|
|
secret = pyotp.random_base32()
|
|
store.provision_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_secret=secret,
|
|
project_ids=("dome-performance",),
|
|
)
|
|
now = int(time.time())
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-1",
|
|
domain="dome-performance",
|
|
now=now,
|
|
)
|
|
first = store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_code=pyotp.TOTP(secret).at(now),
|
|
challenge_id="challenge-1",
|
|
audience="https://gewu.example",
|
|
now=now,
|
|
)
|
|
assert first is not None
|
|
assert store.update_owner_grants("project-owner", ("interactive-video",))
|
|
assert (
|
|
store.authenticate_owner_session(
|
|
first.session_token,
|
|
challenge_id="challenge-1",
|
|
audience="https://gewu.example",
|
|
)
|
|
is None
|
|
)
|
|
|
|
second_counter = now // 30 + 1
|
|
second_now = second_counter * 30
|
|
issue_challenge(
|
|
store,
|
|
principal_id="project-owner",
|
|
challenge_id="challenge-2",
|
|
domain="interactive-video",
|
|
now=second_now,
|
|
)
|
|
second = store.authenticate_owner(
|
|
principal_id="project-owner",
|
|
password="correct horse battery staple",
|
|
totp_code=pyotp.TOTP(secret).generate_otp(second_counter),
|
|
challenge_id="challenge-2",
|
|
audience="https://gewu.example",
|
|
now=second_now,
|
|
)
|
|
assert second is not None
|
|
assert store.revoke_owner("project-owner")
|
|
assert (
|
|
store.authenticate_owner_session(
|
|
second.session_token,
|
|
challenge_id="challenge-2",
|
|
audience="https://gewu.example",
|
|
)
|
|
is None
|
|
)
|