atlas-iac/testing/tests/test_hermes_hux_artifact_auth.py

"""HUX-04 artifacts: authorization boundaries and tenant isolation.

Security obligations exercised: a second subject sees 404, never 403, for
every artifact route (SO-33, SO-34); malformed ids never reach a path (SO-43);
a forged owner field inside the tenant directory is still refused; every
denial leaves an audit outcome; the card serves nothing while its flag is off.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import artifacts, audit, contracts, errors, identity, store  # noqa: E402
from hux.server import build_router  # noqa: E402

SCHEMAS = contracts.load_all()
OWNER = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
OTHER = {**OWNER, "X-Hux-Subject": "usr_fedcba9876543210"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])


@pytest.fixture
def router(tmp_path):
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})


def call(router, method, path, body=None, headers=OWNER):
    payload = json.dumps(body).encode() if body is not None else b""
    response = router.dispatch(method, path, headers, payload)
    return response.status, response.body


@pytest.fixture
def artifact(router):
    status, record = call(router, "POST", "/hux/v1/artifacts", {"type": "markdown", "title": "Mine", "content": "secret\n", "conversation_id": "conv_0001abcd"})
    assert status == 201
    return record


def test_second_subject_gets_404_everywhere(router, artifact):
    art = artifact["id"]
    probes = [
        ("GET", f"/hux/v1/artifacts/{art}", None),
        ("GET", f"/hux/v1/artifacts/{art}/versions/1", None),
        ("GET", f"/hux/v1/artifacts/{art}/versions/1/diff", None),
        ("POST", f"/hux/v1/artifacts/{art}/versions", {"content": "x"}),
        ("POST", f"/hux/v1/artifacts/{art}/promote", {"project_id": "prj_0001aaaa"}),
    ]
    for method, path, body in probes:
        status, error = call(router, method, path, body, {**OTHER, "If-Match": "1"})
        assert (status, error["code"]) == (404, "not_found"), path
        assert contracts.validate_record(error, SCHEMAS) == []
    status, body = call(router, "GET", "/hux/v1/artifacts?conversation_id=conv_0001abcd", headers=OTHER)
    assert status == 200 and body["items"] == []
    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_fedcba9876543210", "chat", "router")))
    assert [r["outcome"] for r in rows if r["action"].startswith("artifacts.")].count("not_found") == 5


def test_cross_tenant_lineage_forgery_is_404(router, artifact):
    body = {"type": "markdown", "title": "Derived", "content": "copy", "lineage": {"artifact_id": artifact["id"], "version": 1}}
    status, error = call(router, "POST", "/hux/v1/artifacts", body, OTHER)
    assert (status, error["code"]) == (404, "not_found")
    status, own = call(router, "POST", "/hux/v1/artifacts", {**body, "lineage": None}, OTHER)
    assert status == 201
    status, error = call(router, "POST", f"/hux/v1/artifacts/{own['id']}/versions", {"content": "v2", "lineage": {"artifact_id": artifact["id"], "version": 1}}, {**OTHER, "If-Match": "1"})
    assert status == 404
    assert call(router, "GET", f"/hux/v1/artifacts/{own['id']}", headers=OTHER)[1]["current_version"] == 1


def test_malformed_ids_never_reach_the_filesystem(router):
    for path in ("/hux/v1/artifacts/..", "/hux/v1/artifacts/art_x", "/hux/v1/artifacts/ART_0000000000"):
        status, error = call(router, "GET", path)
        assert status == 404 and error["code"] == "not_found", path
    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
    assert not (tenant.root / "artifacts" / "...json").exists()


def test_forged_owner_inside_tenant_directory_is_refused(router, artifact):
    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
    forged = {**tenant.get("artifacts", artifact["id"]), "id": "art_forged000001", "owner": "usr_fedcba9876543210"}
    tenant.put("artifacts", forged)
    assert call(router, "GET", "/hux/v1/artifacts/art_forged000001")[0] == 404
    status, error = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c", "lineage": {"artifact_id": "art_forged000001", "version": 1}})
    assert status == 404
    status, body = call(router, "GET", "/hux/v1/artifacts")
    assert [a["id"] for a in body["items"]] == [artifact["id"]]
    assert artifacts.artifact_exists(tenant, "art_forged000001")


def test_contract_validation_guards_every_write(router, artifact):
    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
    current = tenant.get("artifacts", artifact["id"])
    with pytest.raises(errors.Invalid) as raised:
        artifacts.append_version(tenant, current, {"version": 2, "created_at": "now", "created_by": {"type": "user", "id": "x"}, "content_ref": {"hash": "sha256:" + "a" * 64, "bytes": 1, "mime": "x"}}, 1)
    assert raised.value.details and tenant.get("artifacts", artifact["id"])["current_version"] == 1


def test_flag_off_hides_the_card(tmp_path, artifact):
    router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation", "HUX_ROUTER_KEY": "rk"})
    status, error = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c"})
    assert (status, error["code"]) == (404, "flag_off")
    status, error = call(router, "GET", "/hux/v1/artifacts")
    assert status == 404


def test_idempotency_replay_is_scoped_to_the_caller(router):
    body = {"type": "code", "title": "t", "content": "c"}
    status, mine = call(router, "POST", "/hux/v1/artifacts", body, {**OWNER, "Idempotency-Key": "shared-key-01"})
    status, theirs = call(router, "POST", "/hux/v1/artifacts", body, {**OTHER, "Idempotency-Key": "shared-key-01"})
    assert status == 201 and theirs["id"] != mine["id"] and theirs["owner"] == "usr_fedcba9876543210"
    status, again = call(router, "POST", "/hux/v1/artifacts", body, {**OWNER, "Idempotency-Key": "shared-key-02"})
    assert status == 201 and again["id"] != mine["id"]
    status, error = call(router, "POST", f"/hux/v1/artifacts/{mine['id']}/versions", {"content": "v"}, {**OWNER, "If-Match": "1", "Idempotency-Key": "shared-key-02"})
    assert status == 201


def test_audit_rows_name_family_verbs(router, artifact):
    call(router, "GET", f"/hux/v1/artifacts/{artifact['id']}")
    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")))
    assert [r["action"] for r in rows] == ["artifacts.create", "artifacts.get"]
    assert all(contracts.validate("common.schema.json", r, SCHEMAS, "/$defs/audit_outcome") == [] for r in rows)


def test_secret_bearing_content_is_kept_but_forced_restricted(router):
    """Review a2-C / SO-12: user artifacts are not rewritten, but a secret pattern forces restricted sensitivity and is audited."""
    token = "ghp_" + "A" * 36
    status, record = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "env", "content": f"TOKEN={token}", "sensitivity": "personal"})
    assert status == 201 and record["sensitivity"] == "restricted"
    status, version = call(router, "GET", f"/hux/v1/artifacts/{record['id']}/versions/1")
    assert status == 200 and token in version["content"]
    clean = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "clean", "content": "x = 1"})[1]
    assert clean["sensitivity"] == "personal"
    status, bumped = call(router, "POST", f"/hux/v1/artifacts/{clean['id']}/versions", {"content": f"key: {token}"}, {**OWNER, "If-Match": "1"})
    assert status == 201 and bumped["sensitivity"] == "restricted"
    from hux import audit, store
    reasons = [r.get("reason") for r in audit.recent(store.TenantStore(router.data_root, identity.resolve(OWNER, {"HUX_ROUTER_KEY": "rk"}))) if r["action"].startswith("artifacts.")]
    assert reasons.count("secret_pattern_restricted") == 2