canned-prompts/service/tests/test_publisher_identity.py

"""Publisher identity (CANP-WP-0006 / RCP-WP-0002-T05).

App-local publisher tokens per DR-3. The tests that matter here are the ones
about what a credential *cannot* do.
"""

from __future__ import annotations

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from canned_prompts_service.api import create_app
from canned_prompts_service.auth import hash_token, mint_token, owns_namespace
from canned_prompts_service.db import make_engine
from canned_prompts_service.models import NamespaceClaim, Publisher
from canned_prompts_service.settings import Settings

REPO = Path(__file__).resolve().parents[2]
OPERATOR = "operator-token"
OP = {"Authorization": f"Bearer {OPERATOR}"}


def payload(name: str = "house-style", registry: str = "local") -> dict:
    import canned_prompts as cp

    src = REPO / "examples" / name
    manifest = cp.validate_package(src)
    return {
        "registry": registry,
        "source": f"examples/{name}",
        "files": {
            rel: {"text": (src / rel).read_text(encoding="utf-8")}
            for rel in sorted(cp.package_members(src, manifest))
        },
    }


@pytest.fixture()
def app(db_url: str) -> TestClient:
    return TestClient(
        create_app(Settings(database_url=db_url, publish_token=OPERATOR), make_engine(db_url))
    )


# --- token handling ---

def test_tokens_are_unguessable_and_prefixed() -> None:
    a, b = mint_token(), mint_token()
    assert a != b and a.startswith("cp_") and len(a) > 40


def test_tokens_are_stored_hashed(app: TestClient) -> None:
    """A registry that can print its own credentials back can impersonate."""
    token = app.post("/publishers", json={"name": "ada"}, headers=OP).json()["token"]
    listed = app.get("/publishers", headers=OP).json()["publishers"][0]
    assert "token" not in listed
    assert hash_token(token) != token


def test_token_is_shown_once_only(app: TestClient) -> None:
    app.post("/publishers", json={"name": "ada"}, headers=OP)
    body = app.get("/publishers", headers=OP).json()
    assert all("token" not in p for p in body["publishers"])


# --- what a credential cannot do ---

def test_publisher_token_cannot_mint_publishers(app: TestClient) -> None:
    """A publisher that could mint publishers is an administrator in disguise."""
    token = app.post("/publishers", json={"name": "ada"}, headers=OP).json()["token"]
    response = app.post(
        "/publishers", json={"name": "mallory"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 403


def test_unknown_and_wrong_tokens_are_indistinguishable(app: TestClient) -> None:
    """Different answers would let a caller enumerate which tokens exist."""
    a = app.post("/packages", json=payload(), headers={"Authorization": "Bearer cp_nope"})
    b = app.post("/packages", json=payload(), headers={"Authorization": "Bearer garbage"})
    assert a.status_code == b.status_code == 403
    assert a.json()["detail"] == b.json()["detail"]


def test_revoked_publisher_is_refused(app: TestClient) -> None:
    token = app.post("/publishers", json={"name": "ada"}, headers=OP).json()["token"]
    assert app.delete("/publishers/ada", headers=OP).status_code == 200
    response = app.post("/packages", json=payload(), headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403
    assert "revoked" in response.json()["detail"]


def test_revocation_preserves_attribution(app: TestClient) -> None:
    """What someone published stays theirs after their credential is revoked."""
    token = app.post("/publishers", json={"name": "ada"}, headers=OP).json()["token"]
    app.post("/packages", json=payload(), headers={"Authorization": f"Bearer {token}"})
    app.delete("/publishers/ada", headers=OP)
    entry = app.get("/index").json()["entries"][0]
    assert entry["published_by"] == "ada"


# --- namespace ownership (§ 20.1), now enforceable ---

def test_owner_may_publish_into_their_closed_namespace(app: TestClient, session) -> None:
    token = app.post("/publishers", json={"name": "ada"}, headers=OP).json()["token"]
    session.add(NamespaceClaim(tenant="default", registry="local",
                               namespace="practice", policy="closed", owner="ada"))
    session.commit()
    response = app.post("/packages", json=payload(), headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 201


def test_non_owner_is_refused_a_closed_namespace(app: TestClient, session) -> None:
    token = app.post("/publishers", json={"name": "mallory"}, headers=OP).json()["token"]
    session.add(NamespaceClaim(tenant="default", registry="local",
                               namespace="practice", policy="closed", owner="ada"))
    session.commit()
    response = app.post("/packages", json=payload(), headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403
    assert "owned by ada" in response.json()["detail"]


def test_closed_namespace_without_an_owner_admits_nobody() -> None:
    """A missing owner means nobody has been granted it — not that anyone may."""
    claim = NamespaceClaim(tenant="t", registry="r", namespace="n", policy="closed", owner=None)
    assert owns_namespace(claim, "ada", Settings()) is False
    assert owns_namespace(claim, "operator", Settings()) is False


def test_open_and_unclaimed_namespaces_admit_any_publisher() -> None:
    assert owns_namespace(None, "ada", Settings()) is True
    claim = NamespaceClaim(tenant="t", registry="r", namespace="n", policy="open")
    assert owns_namespace(claim, "ada", Settings()) is True


def test_operator_publishes_but_owns_no_closed_namespace(app: TestClient, session) -> None:
    """The operator token is a bootstrap path, not an identity that owns things."""
    session.add(NamespaceClaim(tenant="default", registry="local",
                               namespace="practice", policy="closed", owner="ada"))
    session.commit()
    assert app.post("/packages", json=payload(), headers=OP).status_code == 403