mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
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:",
|
|
):
|
|
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: false",
|
|
):
|
|
self.assertIn(required, deployment)
|
|
self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$")
|
|
|
|
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_agones_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)
|
|
|
|
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_create(self):
|
|
rbac = self.read("rbac.yaml")
|
|
self.assertIn("namespace: agones-system", rbac)
|
|
self.assertIn('resources: ["gameserverallocations"]', rbac)
|
|
self.assertIn('verbs: ["create"]', rbac)
|
|
self.assertNotRegex(rbac, r"verbs:.*\b(get|list|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()
|