audit-core/tests/test_networkpolicies.py

from pathlib import Path


ROOT = Path(__file__).parents[1]


def test_whitehat_ingress_is_bound_to_namespace_and_target_labels():
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-whitehat-ingress" in document
    )

    # Both selectors must remain in the same `from` peer. Splitting them into
    # two list items changes AND to OR and would admit every pod in either set.
    expected_peer = """        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: whitehat
          podSelector:
            matchLabels:
              whitehat.security/plane: \"true\"
              whitehat.security/target: audit-core"""
    assert expected_peer in policy
    assert policy.count("        - namespaceSelector:") == 1
    assert "        - {protocol: TCP, port: 8080}" in policy


def test_approval_engine_ingress_is_bound_to_namespace_and_pod_labels():
    """AUDIT-WP-0009-T09. A load-bearing source gets a narrow rule, not a wide one."""
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-approval-engine-ingress" in document
    )

    # One `from` peer holding both selectors. Two list items would be OR, and
    # would admit every pod in the approval-engine namespace plus every pod
    # anywhere carrying the app label.
    expected_peer = """        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: approval-engine
          podSelector:
            matchLabels:
              app.kubernetes.io/name: approval-engine"""
    assert expected_peer in policy
    assert policy.count("        - namespaceSelector:") == 1
    assert "        - {protocol: TCP, port: 8080}" in policy


def test_user_engine_sender_ingress_is_unchanged_by_the_new_sender():
    """glas-harness asked for user-engine's sender to stay as it is."""
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-sender-ingress" in document
    )
    assert "kubernetes.io/metadata.name: user-engine" in policy
    assert "approval-engine" not in policy


def test_informed_decision_ingress_is_bound_to_namespace_and_pod_labels():
    """AUDIT-WP-0009-T11. A second load-bearing source gets the narrow rule too."""
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-informed-decision-ingress" in document
    )

    expected_peer = """        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: informed-decision
          podSelector:
            matchLabels:
              app.kubernetes.io/name: informed-decision"""
    assert expected_peer in policy
    assert policy.count("        - namespaceSelector:") == 1
    assert "        - {protocol: TCP, port: 8080}" in policy


def test_no_egress_to_informed_decision_is_created():
    """The custody declaration is carried, never dereferenced.

    GH-DEC-2026-014 limit 3 is met by storing an attributable declaration, not
    by audit-core retrieving content. If an egress rule to the emitter ever
    appears, the claim in docs/informed-decision-source-registration.md that
    audit-core performs no retrieval has silently stopped being true.
    """
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    egress = next(
        document for document in documents if "name: audit-core-egress" in document
    )
    assert "informed-decision" not in egress
    assert "approval-engine" not in egress
    # Postgres and DNS only.
    assert egress.count("        - namespaceSelector:") == 2


def test_tenant_engine_ingress_is_bound_to_namespace_and_pod_labels():
    """AUDIT-WP-0010-T03. Attributive is a weaker claim, not a wider rule."""
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-tenant-engine-ingress" in document
    )
    expected_peer = """        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: tenant-engine
          podSelector:
            matchLabels:
              app.kubernetes.io/name: tenant-engine"""
    assert expected_peer in policy
    assert policy.count("        - namespaceSelector:") == 1


def test_every_sender_has_exactly_one_ingress_rule():
    """A fourth sender fails here rather than silently in production.

    AUDIT-WP-0010-T03 asked for this: the policy set and the registered sender
    set must move together, and neither should be able to grow alone.
    """
    import json

    text = (ROOT / "deploy" / "networkpolicies.yaml").read_text()
    scope = json.loads((ROOT / "deploy" / "senders-scope.json").read_text())
    senders = {entry["name"] for entry in scope}
    assert senders == {
        "user-engine", "approval-engine", "informed-decision", "tenant-engine",
        "nk-bounded-executor",
    }
    # user-engine's rule predates the per-sender naming convention.
    assert "name: audit-core-sender-ingress" in text
    for sender in senders - {"user-engine"}:
        assert f"name: audit-core-{sender}-ingress" in text


def test_bounded_executor_ingress_is_bound_to_sso_and_executor_pod_label():
    """AUDIT-WP-0018-T02. Only the executor's pods, not the whole sso namespace.

    Authelia, LLDAP and KeyCape run in sso too; a namespace-only rule would
    let any of them reach the receiver's write port.
    """
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    policy = next(
        document
        for document in documents
        if "name: audit-core-nk-bounded-executor-ingress" in document
    )
    expected_peer = """        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: sso
          podSelector:
            matchLabels:
              app.kubernetes.io/name: bounded-executor"""
    assert expected_peer in policy
    assert policy.count("        - namespaceSelector:") == 1
    assert policy.count("        - {protocol: TCP, port:") == 1
    assert "        - {protocol: TCP, port: 8080}" in policy


def test_no_egress_to_sso_is_created():
    """The receiver answers the executor; it never calls into sso."""
    documents = (ROOT / "deploy" / "networkpolicies.yaml").read_text().split("\n---\n")
    for document in documents:
        if "Egress" in document and "egress:" in document:
            assert "kubernetes.io/metadata.name: sso" not in document