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", "--readiness-max-stale=30s", ): self.assertIn(required, deployment) self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$") def test_maintenance_runs_the_live_abandonment_reconciler_hardened(self): deployment = self.read("maintenance-deployment.yaml") for required in ( "replicas: 2", "type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1", "serviceAccountName: maintenance", "automountServiceAccountToken: false", "runAsNonRoot: true", "type: RuntimeDefault", "allowPrivilegeEscalation: false", "readOnlyRootFilesystem: true", "drop: [ALL]", "image: ghcr.io/cosmic-clash/maintenance@sha256:", "--initial-connect-interval=1s", "--live-abandonment-batch=100", "name: COSMIC_CLASH_POSTGRES_DSN", "key: dsn", "topologySpreadConstraints:", "topologyKey: topology.kubernetes.io/zone", "podAntiAffinity:", "topologyKey: kubernetes.io/hostname", ): self.assertIn(required, deployment) self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$") def test_maintenance_pdb_keeps_one_reconciler_running(self): pdb = self.read("maintenance-pdb.yaml") for required in ( "apiVersion: policy/v1", "kind: PodDisruptionBudget", "name: maintenance", "namespace: cosmic-clash", "minAvailable: 1", "app.kubernetes.io/name: maintenance", ): self.assertIn(required, pdb) 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: /readyz", "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: /readyz", "path: /healthz", "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) def test_maintenance_network_policy_only_allows_postgres_and_dns(self): policies = self.read("network-policies.yaml") maintenance = policies.split("name: maintenance-allowed-egress", 1)[-1] self.assertIn("app.kubernetes.io/name: maintenance", maintenance) for port in ("port: 5432", "port: 53"): self.assertIn(port, maintenance) self.assertNotIn("port: 8080", maintenance) def test_every_required_workload_role_is_deployed(self): # The base deployed a control-plane image nothing built, and built a # matcher image nothing deployed -- so applying it produced a cluster # where tickets could be created but never consumed. Assert the # advertised topology is actually complete. kustomization = self.read("kustomization.yaml") rendered = "".join( self.read(name.strip("- ").strip()) for name in kustomization.splitlines() if name.strip().startswith("- ") and name.strip().endswith(".yaml") ) for role in ("control-plane", "allocator", "maintenance", "matcher"): self.assertIn(f"app.kubernetes.io/name: {role}", rendered, role) self.assertIn("kind: Fleet", rendered) # Casual and ranked must both be scheduled; one matcher process serves # exactly one playlist. self.assertIn("name: matcher-casual", rendered) self.assertIn("name: matcher-ranked", rendered) self.assertIn("--playlist=casual", rendered) self.assertIn("--playlist=ranked", rendered) # Ranked is strictly 3v3; AllocateAcceptedProposal rejects anything else. self.assertIn("--size=6", rendered) def test_every_referenced_image_maps_to_a_real_dockerfile_target(self): dockerfile = (Path(__file__).parents[2] / "Dockerfile").read_text() targets = set(re.findall(r"(?mi)^FROM\s+.*?\bAS\s+(\S+)\s*$", dockerfile)) # Guard the guard: if the target regex stops matching, every image # below would "pass" vacuously. self.assertIn("server", targets) referenced = set() for path in sorted(BASE.glob("*.yaml")): for image in re.findall(r"image:\s*ghcr\.io/cosmic-clash/([\w.-]+)@", path.read_text()): referenced.add(image) self.assertTrue(referenced, "no images were found to check") self.assertEqual(set(), referenced - targets, "manifests reference images this repo cannot build") # testkit-api injects a fake login provider that accepts any ticket. self.assertNotIn("testkit-api", referenced) def test_game_traffic_and_workload_callbacks_are_permitted(self): policies = self.read("network-policies.yaml") # Public players reach the allocated server directly over UDP; the # namespace-wide default deny blocked that entirely. ingress = policies.split("name: game-server-allowed-ingress", 1) self.assertEqual(len(ingress), 2, "game-server ingress policy is missing") game_ingress = ingress[1] self.assertIn("app.kubernetes.io/name: game-server", game_ingress) self.assertIn("protocol: UDP", game_ingress) self.assertIn("port: 7777", game_ingress) # Game servers are control-plane clients: roster fetch, registration, # connection receipts, shutdown and result submission. Their egress was # allowed but the matching control-plane ingress was not. control_plane = policies.split("name: control-plane-allowed-flows", 1)[-1].split("---", 1)[0] self.assertIn("app.kubernetes.io/name: game-server", control_plane) self.assertIn("app.kubernetes.io/name: edge-gateway", control_plane) # The default deny must survive all of this. self.assertIn("name: default-deny-ingress-egress", policies) def test_matcher_network_policy_only_allows_its_datastores_and_dns(self): policies = self.read("network-policies.yaml") matcher = policies.split("name: matcher-allowed-egress", 1)[-1] self.assertIn("app.kubernetes.io/name: matcher", matcher) for port in ("port: 5432", "port: 6379", "port: 53"): self.assertIn(port, matcher) # The matcher never calls the control plane's API. self.assertNotIn("port: 8080", matcher) if __name__ == "__main__": unittest.main()