CosmicClash/server/security/test_kubernetes_policies.py

from pathlib import Path
import re
import unittest


BASE = Path(__file__).parents[2] / "deploy" / "k8s" / "base"


class KubernetesPolicyTest(unittest.TestCase):
    def read(self, name):
        return (BASE / name).read_text()

    def test_namespace_enforces_restricted_pod_security(self):
        namespace = self.read("namespace.yaml")
        for key in ("enforce", "audit", "warn"):
            self.assertIn(f"pod-security.kubernetes.io/{key}: restricted", namespace)

    def test_workload_is_non_root_immutable_and_unprivileged(self):
        deployment = self.read("control-plane-deployment.yaml")
        for required in (
            "runAsNonRoot: true", "type: RuntimeDefault", "allowPrivilegeEscalation: false",
            "readOnlyRootFilesystem: true", "drop: [ALL]", "resources:",
            "image: ghcr.io/cosmic-clash/control-plane@sha256:",
            "--rate-limit=120", "--rate-limit-window=1m", "--rate-limit-max-keys=10000",
            "--trusted-proxy-cidrs=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16,fc00::/7",
            "name: COSMIC_CLASH_POSTGRES_DSN", "key: dsn",
            "name: COSMIC_CLASH_WORKLOAD_SECRET", "key: secret",
        ):
            self.assertIn(required, deployment)
        self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$")
        self.assertIn("secretKeyRef:", deployment)

    def test_allocator_is_hardened_and_uses_only_external_secrets(self):
        deployment = self.read("allocator-deployment.yaml")
        for required in (
            "runAsNonRoot: true", "type: RuntimeDefault", "allowPrivilegeEscalation: false",
            "readOnlyRootFilesystem: true", "drop: [ALL]", "resources:",
            "image: ghcr.io/cosmic-clash/allocator@sha256:",
            "--metrics-addr=:9091", "containerPort: 9091",
            "key: dsn", "key: secret", "automountServiceAccountToken: true",
            "--agones-url=https://kubernetes.default.svc", "--provider-timeout=10s",
        ):
            self.assertIn(required, deployment)
        self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$")

    def test_control_plane_has_health_rollout_and_failure_domain_guards(self):
        deployment = self.read("control-plane-deployment.yaml")
        for required in (
            "readinessProbe:", "livenessProbe:", "path: /healthz", "port: http",
            "type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1",
            "terminationGracePeriodSeconds: 10",
            "topologySpreadConstraints:", "maxSkew: 1",
            "topologyKey: topology.kubernetes.io/zone",
            "whenUnsatisfiable: ScheduleAnyway", "podAntiAffinity:",
            "preferredDuringSchedulingIgnoredDuringExecution:",
            "topologyKey: kubernetes.io/hostname",
        ):
            self.assertIn(required, deployment)

    def test_control_plane_pdb_preserves_one_replica_during_voluntary_disruption(self):
        pdb = self.read("control-plane-pdb.yaml")
        for required in (
            "apiVersion: policy/v1", "kind: PodDisruptionBudget",
            "name: control-plane", "namespace: cosmic-clash",
            "minAvailable: 1", "app.kubernetes.io/name: control-plane",
        ):
            self.assertIn(required, pdb)

    def test_allocator_rollout_keeps_capacity_and_has_health_probes(self):
        deployment = self.read("allocator-deployment.yaml")
        for required in (
            "type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1",
            "terminationGracePeriodSeconds: 10",
            "readinessProbe:", "livenessProbe:",
            "path: /metrics", "port: metrics",
        ):
            self.assertIn(required, deployment)

    def test_allocator_replicas_prefer_separate_failure_domains(self):
        deployment = self.read("allocator-deployment.yaml")
        for required in (
            "topologySpreadConstraints:", "maxSkew: 1",
            "topologyKey: topology.kubernetes.io/zone",
            "whenUnsatisfiable: ScheduleAnyway",
            "podAntiAffinity:", "preferredDuringSchedulingIgnoredDuringExecution:",
            "topologyKey: kubernetes.io/hostname",
        ):
            self.assertIn(required, deployment)
        self.assertGreaterEqual(deployment.count("app.kubernetes.io/name: allocator"), 4)

    def test_allocator_network_policy_has_only_metrics_data_kubernetes_api_and_dns_flows(self):
        policies = self.read("network-policies.yaml")
        allocator = policies.split("name: allocator-allowed-flows", 1)[-1]
        self.assertIn("port: 9091", allocator)
        for port in ("port: 5432", "port: 443", "port: 53"):
            self.assertIn(port, allocator)
        self.assertNotIn("port: 8080", allocator)
        self.assertNotIn("ipBlock:", allocator)
        self.assertNotIn("agones-system", allocator)

    def test_allocator_pdb_preserves_one_replica_during_voluntary_disruption(self):
        pdb = self.read("allocator-pdb.yaml")
        for required in (
            "apiVersion: policy/v1", "kind: PodDisruptionBudget",
            "name: allocator", "namespace: cosmic-clash",
            "minAvailable: 1", "app.kubernetes.io/name: allocator",
        ):
            self.assertIn(required, pdb)

    def test_rbac_is_scoped_to_allocator_agones_operations(self):
        rbac = self.read("rbac.yaml")
        self.assertNotIn("namespace: agones-system", rbac)
        self.assertGreaterEqual(rbac.count("namespace: cosmic-clash"), 3)
        self.assertIn('resources: ["gameservers"]', rbac)
        self.assertIn('verbs: ["list"]', rbac)
        self.assertIn('resources: ["gameserverallocations"]', rbac)
        self.assertIn('verbs: ["create"]', rbac)
        self.assertIn("name: allocator", rbac)
        self.assertNotRegex(rbac, r"verbs:.*\b(watch|update|patch|delete|\*)\b")
        self.assertNotIn('resources: ["*"]', rbac)

    def test_default_deny_and_only_declared_data_dns_edge_flows_exist(self):
        policies = self.read("network-policies.yaml")
        self.assertIn("name: default-deny-ingress-egress", policies)
        self.assertIn("policyTypes: [Ingress, Egress]", policies)
        for port in ("port: 8080", "port: 5432", "port: 6379", "port: 443", "port: 53"):
            self.assertIn(port, policies)
        self.assertNotIn("ipBlock:", policies)


if __name__ == "__main__":
    unittest.main()