Open ports vs used ports on ambient

export SECRETS_FILE=~/path/to/secrets.sh   # exports SOLO_ISTIO_LICENSE_KEY
./scripts/setup-cluster.sh

kubectl --context kind-port-audit apply -f yaml/10-app/
kubectl --context kind-port-audit -n port-audit get pods -o wide

kubectl --context kind-port-audit apply -f yaml/20-policy/

ZT=$(kubectl --context kind-port-audit -n istio-system get pods -l app=ztunnel -o name | head -1)
kubectl --context kind-port-audit -n istio-system logs "$ZT" --since=30s \
  | jq -c 'select(.scope == "access" and .direction == "inbound" and .["dst.namespace"] == "port-audit")' \
  | head -3

{
  "level": "info",
  "time": "2026-07-16T16:43:49.050106Z",
  "scope": "access",
  "message": "connection complete",
  "src.addr": "10.244.2.12:42920",
  "src.workload": "svc-a-64ff6cdff6-lhhv2",
  "src.namespace": "port-audit",
  "src.identity": "spiffe://port-audit/ns/port-audit/sa/svc-a",
  "dst.addr": "10.244.1.6:15008",
  "dst.hbone_addr": "10.244.1.6:8080",
  "dst.service": "svc-b.port-audit.svc.cluster.local",
  "dst.workload": "svc-b-6b5d6f9886-nk2h6",
  "dst.namespace": "port-audit",
  "dst.identity": "spiffe://port-audit/ns/port-audit/sa/default",
  "direction": "inbound",
  "bytes_sent": 205,
  "bytes_recv": 74,
  "duration": "0ms"
}

kubectl --context kind-port-audit -n port-audit exec deploy/svc-a -- \
  curl -s --max-time 3 http://svc-b:7070/ ; echo "exit=$?"

{
  "level": "error",
  "time": "2026-07-16T16:43:43.794875Z",
  "scope": "access",
  "message": "connection complete",
  "src.addr": "10.244.2.12:42920",
  "src.workload": "svc-a-64ff6cdff6-lhhv2",
  "src.namespace": "port-audit",
  "src.identity": "spiffe://port-audit/ns/port-audit/sa/svc-a",
  "dst.addr": "10.244.1.6:15008",
  "dst.hbone_addr": "10.244.1.6:7070",
  "dst.service": "svc-b.port-audit.svc.cluster.local",
  "dst.workload": "svc-b-6b5d6f9886-nk2h6",
  "dst.namespace": "port-audit",
  "dst.identity": "spiffe://port-audit/ns/port-audit/sa/default",
  "direction": "inbound",
  "bytes_sent": 0,
  "bytes_recv": 0,
  "duration": "0ms",
  "error": "connection closed due to policy rejection: allow policies exist, but none allowed"
}

./scripts/build-collector.sh    # docker build collector.py + kind load
kubectl --context kind-port-audit apply -f yaml/30-audit/
kubectl --context kind-port-audit -n port-audit-system get pods

#!/usr/bin/env python3
"""collector.py - one per node (DaemonSet).

Streams the LOCAL ztunnel's JSON access logs over a single follow=true log
connection and merge-patches this node's key in the shared port-audit-report
ConfigMap. Compared to the polling shell version this replaced, every log
line is handled exactly once (no overlapping --since windows), the API server
holds one long-lived connection instead of a request every 30 seconds, and
the ConfigMap is only written when a port set actually changes (plus a
heartbeat so the key's `updated` stamp stays live).

Stdlib only, on purpose: the Kubernetes API is plain HTTPS with a
ServiceAccount bearer token, so the image needs no pip installs.

Classification is the report's whole taxonomy, unchanged from the shell
version:
  completed without error          -> used
  error contains "policy rejection"-> denied
  any other error (e.g. refused)   -> ignored (stays in the unused column)
Entries without src.identity are dropped (kubelet probes are not usage).
"""

import base64
import gzip
import json
import os
import socket
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request

API = "https://kubernetes.default.svc"
SA = "/var/run/secrets/kubernetes.io/serviceaccount"

NODE = os.environ["NODE_NAME"]
NS_APP = os.environ.get("NS_APP", "port-audit")
NS_AUDIT = os.environ.get("NS_AUDIT", "port-audit-system")
CM = os.environ.get("REPORT_CM", "port-audit-report")
ZT_NS = os.environ.get("ZTUNNEL_NAMESPACE", "istio-system")
# Debounce between change-driven patches, and the idle heartbeat interval.
DEBOUNCE = float(os.environ.get("PATCH_DEBOUNCE_SECONDS", "2"))
HEARTBEAT = int(os.environ.get("HEARTBEAT_SECONDS", "60"))

SSL_CTX = ssl.create_default_context(cafile=f"{SA}/ca.crt")


def log(msg):
    print(time.strftime("%H:%M:%S", time.gmtime()), msg, flush=True)


def request(method, path, body=None, content_type="application/json", timeout=15):
    # Re-read the token per request: projected SA tokens rotate.
    with open(f"{SA}/token") as f:
        token = f.read().strip()
    req = urllib.request.Request(API + path, data=body, method=method)
    req.add_header("Authorization", f"Bearer {token}")
    if body is not None:
        req.add_header("Content-Type", content_type)
    return urllib.request.urlopen(req, context=SSL_CTX, timeout=timeout)


def find_ztunnel():
    query = urllib.parse.urlencode({
        "labelSelector": "app=ztunnel",
        "fieldSelector": f"spec.nodeName={NODE},status.phase=Running",
    })
    with request("GET", f"/api/v1/namespaces/{ZT_NS}/pods?{query}") as resp:
        items = json.load(resp).get("items", [])
    return items[0]["metadata"]["name"] if items else None


def pack(state):
    """Serialise a node key as gzip+base64. One ConfigMap has a hard 1 MiB
    budget shared across ALL its keys (the API server rejects the whole object
    past 1,048,576 bytes), and port/pod sets compress heavily, so a compressed
    key keeps a large fleet under the cap. base64 makes the gzip bytes a valid
    UTF-8 string so it can live in `data` (not `binaryData`) and be merge-patched
    per key like before."""
    raw = json.dumps(state, sort_keys=True).encode()
    return base64.b64encode(gzip.compress(raw)).decode()


def unpack(value):
    """Inverse of pack(); tolerates a pre-compression plain-JSON key."""
    try:
        return json.loads(gzip.decompress(base64.b64decode(value)))
    except (ValueError, OSError):
        return json.loads(value)


def seed_state():
    # A restart keeps the history this node already reported.
    try:
        with request("GET", f"/api/v1/namespaces/{NS_AUDIT}/configmaps/{CM}") as resp:
            data = json.load(resp).get("data") or {}
        if NODE in data:
            return unpack(data[NODE])
    except (urllib.error.URLError, ValueError) as exc:
        log(f"seed skipped: {exc}")
    return {"node": NODE, "services": {}, "pods": {}, "denied": {}}


def patch_state(state):
    # A JSON merge patch that touches ONLY this node's key: applied
    # server-side and atomically, so concurrent writers on other nodes can
    # never be clobbered and no resourceVersion retry loop is needed. The value
    # is gzip+base64 (see pack()) to stay inside the ConfigMap's 1 MiB budget.
    state["updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    body = json.dumps({"data": {NODE: pack(state)}}).encode()
    with request("PATCH", f"/api/v1/namespaces/{NS_AUDIT}/configmaps/{CM}",
                 body=body, content_type="application/merge-patch+json") as resp:
        resp.read()


def classify(entry):
    """Return ('used'|'denied', service, workload, port) or None."""
    if entry.get("scope") != "access" or entry.get("direction") != "inbound":
        return None
    if entry.get("dst.namespace") != NS_APP or not entry.get("src.identity"):
        return None
    addr = entry.get("dst.hbone_addr") or entry.get("dst.addr") or ""
    try:
        port = int(addr.rsplit(":", 1)[1])
    except (IndexError, ValueError):
        return None
    error = entry.get("error") or ""
    if "policy rejection" in error:
        kind = "denied"
    elif error:
        return None  # refused/reset etc: NOT used, NOT denied
    else:
        kind = "used"
    return kind, entry.get("dst.service") or "unattributed", \
        entry.get("dst.workload") or "unattributed", port


def record(state, kind, service, workload, port):
    """Union the observation into the state; return True if anything changed."""
    changed = False
    if kind == "denied":
        buckets = [(state["denied"], service)]
    else:
        buckets = [(state["services"], service), (state["pods"], workload)]
    for mapping, key in buckets:
        ports = mapping.setdefault(key, [])
        if port not in ports:
            ports.append(port)
            ports.sort()
            changed = True
    return changed


def stream(ztunnel, state):
    """Follow the ztunnel log stream; patch on change (debounced) and on
    heartbeat. The read timeout doubles as the heartbeat timer: a quiet
    stream raises timeout, we patch if needed, and the caller reconnects
    with a sinceSeconds overlap (sets make the overlap harmless)."""
    query = urllib.parse.urlencode({"follow": "true", "sinceSeconds": str(HEARTBEAT)})
    resp = request("GET", f"/api/v1/namespaces/{ZT_NS}/pods/{ztunnel}/log?{query}",
                   timeout=HEARTBEAT)
    log(f"streaming logs from {ztunnel}")
    last_patch = 0.0
    dirty = False
    with resp:
        for raw in resp:
            try:
                entry = json.loads(raw)
            except ValueError:
                continue
            if isinstance(entry, dict):
                hit = classify(entry)
                if hit and record(state, *hit):
                    dirty = True
            now = time.time()
            if (dirty and now - last_patch >= DEBOUNCE) or now - last_patch >= HEARTBEAT:
                patch_state(state)
                last_patch = now
                dirty = False
                log(f"patched {CM} key={NODE}")


def main():
    state = seed_state()
    patch_state(state)  # make the key exist (and stamp `updated`) immediately
    log(f"collector up on {NODE}")
    while True:
        try:
            ztunnel = find_ztunnel()
            if not ztunnel:
                log(f"no running ztunnel on {NODE} yet")
                time.sleep(5)
                continue
            stream(ztunnel, state)
            log("log stream ended (ztunnel rotated?), reconnecting")
        except (urllib.error.URLError, socket.timeout, TimeoutError, OSError) as exc:
            # Idle heartbeat lands here too: a quiet stream times out.
            try:
                patch_state(state)
            except (urllib.error.URLError, OSError) as patch_exc:
                log(f"heartbeat patch failed: {patch_exc}")
            log(f"stream interrupted ({exc.__class__.__name__}), reconnecting")
        time.sleep(2)


if __name__ == "__main__":
    main()

# The per-node collector. Stdlib-only Python (the Kubernetes API is plain
# HTTPS + a ServiceAccount bearer token), so there is no pip layer to build
# or patch. Runs as a non-root user.
FROM python:3.12-alpine
COPY collector.py /collector.py
USER 65532:65532
ENTRYPOINT ["python3", "-u", "/collector.py"]

#!/usr/bin/env python3
"""aggregate.py - runs as a CronJob, once a minute.

Merges every node's observations from the report ConfigMap, reads the
CONFIGURED surface (Service ports + AuthorizationPolicy allowed ports), and
writes the diff to the report.json key of the same ConfigMap: per service,
what is exposed, what is allowed, what is actually used, what is not, and
what got denied.

Same image and same stdlib-only style as collector.py; the two scripts are
the whole audit.
"""

import base64
import gzip
import json
import os
import ssl
import time
import urllib.request

API = "https://kubernetes.default.svc"
SA = "/var/run/secrets/kubernetes.io/serviceaccount"

NS_APP = os.environ.get("NS_APP", "port-audit")
NS_AUDIT = os.environ.get("NS_AUDIT", "port-audit-system")
CM = os.environ.get("REPORT_CM", "port-audit-report")

SSL_CTX = ssl.create_default_context(cafile=f"{SA}/ca.crt")


def request(method, path, body=None, content_type="application/json"):
    with open(f"{SA}/token") as f:
        token = f.read().strip()
    req = urllib.request.Request(API + path, data=body, method=method)
    req.add_header("Authorization", f"Bearer {token}")
    if body is not None:
        req.add_header("Content-Type", content_type)
    with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
        return json.load(resp)


def unpack(value):
    """Node keys are written gzip+base64 by the collector (to stay inside the
    ConfigMap's 1 MiB budget); decode them here. Falls back to plain JSON."""
    try:
        return json.loads(gzip.decompress(base64.b64decode(value)))
    except (ValueError, OSError):
        return json.loads(value)


def union(maps):
    """Merge a list of {key: [ports]} maps into one, unioning the port lists."""
    merged = {}
    for mapping in maps:
        for key, ports in mapping.items():
            merged[key] = sorted(set(merged.get(key, [])) | set(ports))
    return merged


def allowed_ports(policies, app):
    """Ports named by ALLOW AuthorizationPolicies whose selector matches this
    app label (the labelling convention this lab uses throughout)."""
    ports = set()
    for policy in policies:
        spec = policy.get("spec", {})
        if spec.get("action", "ALLOW") != "ALLOW":
            continue
        selector = spec.get("selector", {}).get("matchLabels", {})
        if selector.get("app") != app:
            continue
        for rule in spec.get("rules", []):
            for to in rule.get("to", []):
                for port in to.get("operation", {}).get("ports", []):
                    ports.add(int(port))
    return sorted(ports)


def main():
    cm = request("GET", f"/api/v1/namespaces/{NS_AUDIT}/configmaps/{CM}")
    nodes = [unpack(value) for key, value in (cm.get("data") or {}).items()
             if key != "report.json"]
    services = request("GET", f"/api/v1/namespaces/{NS_APP}/services")["items"]
    policies = request(
        "GET", f"/apis/security.istio.io/v1/namespaces/{NS_APP}/authorizationpolicies"
    )["items"]

    observed = union(n.get("services", {}) for n in nodes)
    observed_pods = union(n.get("pods", {}) for n in nodes)
    denied = union(n.get("denied", {}) for n in nodes)

    rows = []
    for svc in services:
        name = svc["metadata"]["name"]
        fqdn = f"{name}.{svc['metadata']['namespace']}.svc.cluster.local"
        configured = sorted(p["port"] for p in svc["spec"].get("ports", []))
        allowed = allowed_ports(policies, svc["spec"].get("selector", {}).get("app"))
        used = observed.get(fqdn, [])
        rows.append({
            "service": name,
            "configured_service_ports": configured,
            "authz_allowed_ports": allowed,
            "used_ports": used,
            "unused_ports": [p for p in configured if p not in used],
            "authz_allowed_never_used": [p for p in allowed if p not in used],
            "denied_attempts": denied.get(fqdn, []),
            "over_provisioned": any(p not in used for p in configured),
        })

    report = {
        "generated": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "nodes_reporting": [n["node"] for n in nodes if "node" in n],
        "services": rows,
        "pods": observed_pods,
    }
    body = json.dumps({"data": {"report.json": json.dumps(report, indent=2)}}).encode()
    request("PATCH", f"/api/v1/namespaces/{NS_AUDIT}/configmaps/{CM}",
            body=body, content_type="application/merge-patch+json")
    print("report.json updated:")
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

kubectl --context kind-port-audit -n port-audit-system get cm port-audit-report -o json \
  | jq -r '.data["port-audit-worker"]' | base64 -d | gunzip | jq .

{
  "node": "port-audit-worker2",
  "services": {
    "svc-b.port-audit.svc.cluster.local": [8080, 8081, 8082, 9090, 9091, 9092]
  },
  "pods": {
    "svc-b-68b48c6658-59qkv": [8080, 8081, 8082, 9090, 9091, 9092]
  },
  "denied": {
    "svc-b.port-audit.svc.cluster.local": [7070]
  },
  "updated": "2026-07-16T17:05:33Z"
}

kubectl --context kind-port-audit -n port-audit-system get cm port-audit-report \
  -o jsonpath='{.data.report\.json}' | jq .

{
  "generated": "2026-07-16T17:05:00Z",
  "nodes_reporting": ["port-audit-worker", "port-audit-worker2"],
  "services": [
    {
      "service": "svc-b",
      "configured_service_ports": [7070, 8080, 8081, 8082, 8083, 8084,
                                   9090, 9091, 9092, 9093, 9094],
      "authz_allowed_ports":      [8080, 8081, 8082, 8083, 8084,
                                   9090, 9091, 9092, 9093, 9094],
      "used_ports":               [8080, 8081, 8082, 9090, 9091, 9092],
      "unused_ports":             [7070, 8083, 8084, 9093, 9094],
      "authz_allowed_never_used": [8083, 8084, 9093, 9094],
      "denied_attempts":          [7070],
      "over_provisioned": true
    }
  ],
  "pods": {
    "svc-b-68b48c6658-wwfhd": [8080, 8081, 8082, 9090, 9091, 9092],
    "svc-b-68b48c6658-59qkv": [8080, 8081, 8082, 9090, 9091, 9092]
  }
}

kubectl --context kind-port-audit apply -f yaml/40-remediate/
# prove it: 8083 was allowed a minute ago, now it is not
kubectl --context kind-port-audit -n port-audit exec deploy/svc-a -- \
  curl -s --max-time 3 http://svc-b:8083/ ; echo "exit=$?"

export ANTHROPIC_API_KEY=...   # the model the agent runs on
export GITHUB_PAT=...          # a PAT with Contents:read+write on the repo
make kagent-setup

make kagent-ui    # http://localhost:8080

Publish the port audit to tjorourke/solo-port-test at port-audit-report.md on main