GEWU/tests/api/test_owner_session.py

import asyncio
import hashlib
import json
import time

import httpx2
import pyotp
from cryptography.fernet import Fernet
from mcp import Client
from mcp.client.streamable_http import streamable_http_client

from gewu.adapters.git_repository import GitCanonicalRepository
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
from gewu.application.reading import CanonicalNodeReader
from gewu.bootstrap.app import create_app


def test_owner_login_returns_a_short_lived_non_agent_token(tmp_path) -> None:
    async def exercise() -> None:
        repository = GitCanonicalRepository(tmp_path / "knowledge.git")
        identities = SQLiteIdentityStore(
            tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
        )
        secret = pyotp.random_base32()
        identities.provision_owner(
            principal_id="owner-one",
            password="correct horse battery staple",
            totp_secret=secret,
            project_ids=("dome-performance",),
        )
        now = int(time.time())
        identities.issue_owner_challenge(
            challenge_id="challenge-one",
            owner_principal_id="owner-one",
            proposal_id="proposal-one",
            proposal_revision=1,
            payload_digest="a" * 64,
            diff_digest="b" * 64,
            domains=("dome-performance",),
            expires_at=now + 300,
            now=now,
        )
        agent_token = identities.provision(
            credential_id="agent-key",
            principal_id="agent-one",
            roles=("agent",),
            project_ids=("dome-performance",),
        )
        app = create_app(
            identity_provider=identities,
            owner_identity_provider=identities,
            owner_service_audience="gewu-local-test",
            node_reader=CanonicalNodeReader(repository),
        )
        async with httpx2.AsyncClient(
            transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1"
        ) as client:
            response = await client.post(
                "/auth/owner-session",
                json={
                    "principal_id": "owner-one",
                    "password": "correct horse battery staple",
                    "totp_code": pyotp.TOTP(secret).now(),
                    "challenge_id": "challenge-one",
                },
            )
            assert response.status_code == 200, response.text
            assert response.headers["cache-control"] == "no-store"
            token = response.json()["session_token"]
            assert token.startswith("gwo_")
            assert identities.authenticate(token) is None
            assert identities.authenticate(agent_token) is not None
            assert (
                identities.authenticate_owner_session(
                    token,
                    challenge_id="challenge-one",
                    audience="gewu-local-test",
                )
                is not None
            )

            wrong_origin = await client.post(
                "/auth/owner-session",
                headers={"Origin": "https://attacker.invalid"},
                json={
                    "principal_id": "owner-one",
                    "password": "correct horse battery staple",
                    "totp_code": pyotp.TOTP(secret).now(),
                    "challenge_id": "challenge-two",
                },
            )
            assert wrong_origin.status_code == 403

            oversized = await client.post(
                "/auth/owner-session",
                content='{"padding":"' + ("x" * 5000) + '"}',
                headers={"Content-Type": "application/json"},
            )
            assert oversized.status_code == 413

        async with httpx2.AsyncClient(
            transport=httpx2.ASGITransport(app=app), base_url="http://gewu.example"
        ) as remote_http:
            insecure = await remote_http.post(
                "/auth/owner-session",
                json={
                    "principal_id": "owner-one",
                    "password": "correct horse battery staple",
                    "totp_code": "000000",
                    "challenge_id": "challenge-three",
                },
            )
            assert insecure.status_code == 403

    asyncio.run(exercise())


def test_owner_login_is_fail_closed_without_encryption_key_or_audience(tmp_path) -> None:
    async def exercise() -> None:
        repository = GitCanonicalRepository(tmp_path / "knowledge.git")
        identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
        app = create_app(
            identity_provider=identities,
            owner_identity_provider=identities,
            node_reader=CanonicalNodeReader(repository),
        )
        async with httpx2.AsyncClient(
            transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1"
        ) as client:
            response = await client.post(
                "/auth/owner-session",
                json={
                    "principal_id": "owner-one",
                    "password": "correct horse battery staple",
                    "totp_code": "000000",
                    "challenge_id": "challenge-one",
                },
            )
            assert response.status_code == 503

    asyncio.run(exercise())


def test_dedicated_owner_mcp_reads_diff_and_records_decision(tmp_path) -> None:
    async def exercise() -> None:
        repository = GitCanonicalRepository(tmp_path / "knowledge.git")
        identities = SQLiteIdentityStore(
            tmp_path / "identity.sqlite3", owner_encryption_key=Fernet.generate_key()
        )
        secret = pyotp.random_base32()
        identities.provision_owner(
            principal_id="owner-one",
            password="correct horse battery staple",
            totp_secret=secret,
            project_ids=("dome-performance",),
        )
        diff_json = '{"files":[{"path":"projects/dome/nodes/a.yaml"}]}'
        diff_digest = hashlib.sha256(diff_json.encode()).hexdigest()
        identities.issue_owner_challenge(
            challenge_id="challenge-review",
            owner_principal_id="owner-one",
            proposal_id="proposal-one",
            proposal_revision=1,
            payload_digest="a" * 64,
            diff_digest=diff_digest,
            domains=("dome-performance",),
            expires_at=int(time.time()) + 300,
            diff_json=diff_json,
        )
        app = create_app(
            identity_provider=identities,
            owner_identity_provider=identities,
            owner_service_audience="gewu-owner-test",
            node_reader=CanonicalNodeReader(repository),
        )
        async with app.router.lifespan_context(app):
            async with httpx2.AsyncClient(
                transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1"
            ) as login_http:
                login = await login_http.post(
                    "/auth/owner-session",
                    json={
                        "principal_id": "owner-one",
                        "password": "correct horse battery staple",
                        "totp_code": pyotp.TOTP(secret).now(),
                        "challenge_id": "challenge-review",
                    },
                )
                assert login.status_code == 200, login.text
                token = login.json()["session_token"]
            async with httpx2.AsyncClient(
                transport=httpx2.ASGITransport(app=app),
                base_url="http://localhost",
                headers={"Authorization": f"Bearer {token}"},
            ) as owner_http:
                async with Client(
                    streamable_http_client("http://localhost/owner-mcp", http_client=owner_http)
                ) as owner:
                    names = {tool.name for tool in (await owner.list_tools()).tools}
                    assert names == {
                        "owner_challenge_get",
                        "owner_diff_get",
                        "owner_decide",
                    }
                    challenge = await owner.call_tool(
                        "owner_challenge_get", {"challenge_id": "challenge-review"}
                    )
                    assert not challenge.is_error
                    diff = await owner.call_tool(
                        "owner_diff_get", {"challenge_id": "challenge-review"}
                    )
                    assert not diff.is_error
                    decision = await owner.call_tool(
                        "owner_decide",
                        {
                            "challenge_id": "challenge-review",
                            "expected_proposal_revision": 1,
                            "expected_payload_digest": "a" * 64,
                            "expected_diff_digest": diff_digest,
                            "decision": "authorize",
                            "rationale": "Reviewed this exact change and approve it.",
                        },
                    )
                    assert not decision.is_error
                    assert json.loads(decision.content[0].text)["status"] == "authorized"

    asyncio.run(exercise())