audit-core/audit_core/attest_publish.py

"""Publish a chain-head attestation where the receiver can read it.

AUDIT-WP-0009-T02. ``attest-chain`` produces an attestation; this puts it
somewhere the running receiver will actually consult, which is the half that
makes `tamper_evidence` able to be true at all.

Two decisions worth stating, because both are refusals.

**It will not publish over a broken chain.** If ``verify_chain`` reports a
break, the job fails and leaves the previous attestation in place. Publishing a
fresh head over a break would replace an honest ``chain_break`` with a
fresh-looking attestation and hide exactly the thing the chain exists to show —
a stale attestation degrades the claim visibly, a false one does not.

**It writes through the API server, not to a shared volume.** The attestation
has to live outside the store it attests, and it has to be written by something
other than the receiver: a receiver that could rewrite its own attestation
could forge it. So this runs as a separate workload with a separate identity
whose Role reaches exactly one named ConfigMap, and the receiver mounts that
ConfigMap read-only with no API-server egress of its own.

The bound this does *not* reach: a cluster owner can rewrite the ConfigMap.
This control defends against a database owner, and `docs/integrity.md` says so.
The logical-offsite copy is the one that survives losing the cluster, and it is
deliberately not written here — audit-core holds no Nextcloud credential and
should not acquire one to publish a hash.
"""

from __future__ import annotations

import json
import os
import ssl
import sys
import urllib.request
from pathlib import Path
from typing import Any

SERVICE_ACCOUNT_DIR = "/var/run/secrets/kubernetes.io/serviceaccount"
CONFIGMAP_NAME = "audit-core-chain-head"
CONFIGMAP_KEY = "chain-head.json"


def _service_account(directory: str) -> tuple[str, str, str]:
    base = Path(directory)
    token = (base / "token").read_text().strip()
    namespace = (base / "namespace").read_text().strip()
    return token, namespace, str(base / "ca.crt")


def publish(
    attestation: dict[str, Any],
    *,
    directory: str = SERVICE_ACCOUNT_DIR,
    host: str | None = None,
    opener: Any = None,
) -> int:
    """PATCH the attestation into the chain-head ConfigMap.

    Returns the HTTP status. Raises rather than swallowing: a failed publish
    must fail the job, because a silent failure leaves a stale attestation
    that will quietly age out of the freshness window with nobody told.
    """
    token, namespace, ca = _service_account(directory)
    api = host or "https://{}:{}".format(
        os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc"),
        os.environ.get("KUBERNETES_SERVICE_PORT", "443"),
    )
    url = f"{api}/api/v1/namespaces/{namespace}/configmaps/{CONFIGMAP_NAME}"
    body = json.dumps({"data": {CONFIGMAP_KEY: json.dumps(attestation, sort_keys=True)}})
    request = urllib.request.Request(
        url,
        data=body.encode(),
        method="PATCH",
        headers={
            "Authorization": f"Bearer {token}",
            # Merge patch, so the job replaces one key rather than the object.
            "Content-Type": "application/merge-patch+json",
            "Accept": "application/json",
        },
    )
    if opener is not None:
        response = opener(request)
    else:
        context = ssl.create_default_context(cafile=ca)
        response = urllib.request.urlopen(request, context=context, timeout=30)
    with response:
        return response.status


def main(argv: list[str] | None = None) -> int:
    from audit_core.cli import _postgres_backend
    from audit_core.integrity import write_attestation

    output = os.environ.get("AUDIT_CORE_ATTESTATION_OUTPUT", "/tmp/chain-head.json")
    schema = os.environ.get("AUDIT_CORE_DATABASE_SCHEMA", "audit_core")

    backend = _postgres_backend(None, schema, migrate=False)
    try:
        report = backend.verify_chain()
        attestation = write_attestation(output, report)
    finally:
        backend.close()

    if not report.intact:
        # Leave the previous attestation standing. A degraded claim is the
        # correct output of a broken chain; a fresh one would be a lie.
        sys.stderr.write(
            "chain not intact ({}); refusing to publish attestation\n".format(
                getattr(report, "first_break", None)
            )
        )
        return 1

    status = publish(attestation)
    print(json.dumps({"published": True, "status": status, **attestation}, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())