83 lines
3 KiB
Python
83 lines
3 KiB
Python
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
CONTRACT = REPO_ROOT / "config/gcp-service-communications.json"
|
|
|
|
|
|
def run_checker(contract: Path, output: Path) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
[
|
|
"python3",
|
|
"ops/check_gcp_service_communications.py",
|
|
"--contract",
|
|
str(contract),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def test_gcp_service_communications_contract_passes(tmp_path: Path) -> None:
|
|
output = tmp_path / "communications.json"
|
|
completed = run_checker(CONTRACT, output)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
payload = json.loads(output.read_text())
|
|
assert payload["status"] == "pass"
|
|
assert payload["allowed_path_count"] == 8
|
|
assert payload["problems"] == []
|
|
|
|
|
|
def test_gcp_service_communications_rejects_broad_ssh(tmp_path: Path) -> None:
|
|
contract = json.loads(CONTRACT.read_text())
|
|
for path in contract["allowed_paths"]:
|
|
if path["name"] == "operator_ssh_to_teleo_vms":
|
|
path["source"]["cidrs"] = ["0.0.0.0/0"]
|
|
contract_path = tmp_path / "bad-communications.json"
|
|
output = tmp_path / "communications.json"
|
|
contract_path.write_text(json.dumps(contract))
|
|
|
|
completed = run_checker(contract_path, output)
|
|
|
|
assert completed.returncode == 1
|
|
payload = json.loads(output.read_text())
|
|
assert payload["status"] == "fail"
|
|
assert any("broad_source_cidr" in problem for problem in payload["problems"])
|
|
|
|
|
|
def test_gcp_service_communications_rejects_public_or_unencrypted_cloudsql(tmp_path: Path) -> None:
|
|
contract = json.loads(CONTRACT.read_text())
|
|
for path in contract["allowed_paths"]:
|
|
if path["name"] == "teleo_vms_to_cloudsql_private":
|
|
path["target"]["public_ip"] = True
|
|
path["encryption"] = "disabled"
|
|
contract_path = tmp_path / "bad-cloudsql.json"
|
|
output = tmp_path / "communications.json"
|
|
contract_path.write_text(json.dumps(contract))
|
|
|
|
completed = run_checker(contract_path, output)
|
|
|
|
assert completed.returncode == 1
|
|
payload = json.loads(output.read_text())
|
|
assert "teleo_vms_to_cloudsql_private:cloudsql_public_ip_not_false" in payload["problems"]
|
|
assert "teleo_vms_to_cloudsql_private:cloudsql_encryption_not_encrypted_only" in payload["problems"]
|
|
|
|
|
|
def test_gcp_service_communications_rejects_secret_value_fields(tmp_path: Path) -> None:
|
|
contract = json.loads(CONTRACT.read_text())
|
|
contract["secret_value"] = "do-not-store-this"
|
|
contract_path = tmp_path / "bad-secret.json"
|
|
output = tmp_path / "communications.json"
|
|
contract_path.write_text(json.dumps(contract))
|
|
|
|
completed = run_checker(contract_path, output)
|
|
|
|
assert completed.returncode == 1
|
|
payload = json.loads(output.read_text())
|
|
assert any("secret_value_fields_present" in problem for problem in payload["problems"])
|