forked from BasedHardware/omi
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_check_deployment_secret_boundary.py
More file actions
181 lines (148 loc) · 7.16 KB
/
Copy pathtest_check_deployment_secret_boundary.py
File metadata and controls
181 lines (148 loc) · 7.16 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
#!/usr/bin/env python3
"""Unit fixtures for the name-only deployment setting boundary ratchet."""
from __future__ import annotations
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path, PurePosixPath, PureWindowsPath
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[2]
CHECKER_PATH = REPO_ROOT / ".github" / "scripts" / "check_deployment_secret_boundary.py"
SPEC = importlib.util.spec_from_file_location("deployment_secret_boundary", CHECKER_PATH)
if SPEC is None or SPEC.loader is None:
raise RuntimeError(f"cannot load {CHECKER_PATH}")
CHECKER = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = CHECKER
SPEC.loader.exec_module(CHECKER)
POLICY = {
"kinds": {
"secret": ["FAKE_SERVER_SECRET"],
"config": ["FAKE_RUNTIME_CONFIG"],
"public_build": ["FAKE_PUBLIC_BUILD"],
},
"exceptions": {},
}
def git_environment() -> dict[str, str]:
"""Temporary fixture repositories must not inherit the hook's Git paths."""
return {name: value for name, value in os.environ.items() if not name.startswith("GIT_")}
class DeploymentSecretBoundaryFixture(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory(prefix="omi-deployment-boundary-")
self.root = Path(self.temp_dir.name)
subprocess.run(["git", "init", "-q"], cwd=self.root, check=True, env=git_environment())
subprocess.run(
["git", "config", "user.email", "test@example.invalid"], cwd=self.root, check=True, env=git_environment()
)
subprocess.run(
["git", "config", "user.name", "Boundary Test"], cwd=self.root, check=True, env=git_environment()
)
self.write("config/deployment-setting-classification.json", json.dumps(POLICY))
self.write(".github/workflows/deploy.yml", "name: test\n# utf-8 guard: \u2603\n")
self.commit("baseline")
def tearDown(self) -> None:
self.temp_dir.cleanup()
def write(self, relative_path: str, contents: str) -> None:
path = self.root / relative_path
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(contents, encoding="utf-8")
def commit(self, message: str) -> None:
subprocess.run(["git", "add", "."], cwd=self.root, check=True, env=git_environment())
subprocess.run(["git", "commit", "-qm", message], cwd=self.root, check=True, env=git_environment())
def errors(self) -> list[str]:
policy = CHECKER.load_policy(self.root / "config/deployment-setting-classification.json")
return CHECKER.validate_policy(policy) + CHECKER.validate_bindings(
policy,
CHECKER.extract_current_bindings(self.root),
CHECKER.extract_base_bindings(self.root, "HEAD"),
)
def test_accepts_correct_secret_and_variable_bindings(self) -> None:
self.write(
".github/workflows/deploy.yml",
"""name: deploy
jobs:
deploy:
env:
CONFIG: ${{ vars.FAKE_RUNTIME_CONFIG }}
TOKEN: ${{ secrets.FAKE_SERVER_SECRET }}
BUILD: ${{ vars.FAKE_PUBLIC_BUILD }}
""",
)
self.assertEqual(self.errors(), [])
def test_rejects_public_build_setting_from_github_secret(self) -> None:
self.write(".github/workflows/deploy.yml", "BUILD: ${{ secrets.FAKE_PUBLIC_BUILD }}\n")
self.assertIn(
"public_build setting FAKE_PUBLIC_BUILD must use vars.FAKE_PUBLIC_BUILD", "\n".join(self.errors())
)
def test_rejects_config_external_secret_mapping(self) -> None:
self.write(
"backend/charts/backend-secrets/dev_omi_backend_secrets_values.yaml",
"""externalSecret:
secretKeys:
- secretKey: FAKE_RUNTIME_CONFIG
remoteKey: FAKE_RUNTIME_CONFIG
""",
)
self.assertIn(
"external_secret binding FAKE_RUNTIME_CONFIG is config; expected secret", "\n".join(self.errors())
)
def test_rejects_secret_from_github_variable(self) -> None:
self.write(".github/workflows/deploy.yml", "TOKEN: ${{ vars.FAKE_SERVER_SECRET }}\n")
self.assertIn(
"github_vars binding FAKE_SERVER_SECRET is secret; expected config or public_build",
"\n".join(self.errors()),
)
def test_rejects_config_loaded_from_secret_manager(self) -> None:
self.write(
".github/workflows/deploy.yml",
'echo "FAKE_RUNTIME_CONFIG=$(gcloud secrets versions access latest --secret=FAKE_SERVER_SECRET)"\n',
)
self.assertIn("secret_manager binding FAKE_RUNTIME_CONFIG is config; expected secret", "\n".join(self.errors()))
def test_rejects_new_unclassified_binding_but_allows_legacy_baseline(self) -> None:
self.write(".github/workflows/deploy.yml", "TOKEN: ${{ secrets.FAKE_LEGACY_NAME }}\n")
self.commit("legacy binding")
self.assertEqual(self.errors(), [])
self.write(
".github/workflows/other.yml",
"TOKEN: ${{ secrets.FAKE_LEGACY_NAME }}\n",
)
self.assertIn("github_secrets binding FAKE_LEGACY_NAME is unclassified", "\n".join(self.errors()))
def test_rejects_malformed_exception_metadata(self) -> None:
policy = dict(POLICY)
policy["exceptions"] = {"FAKE_RUNTIME_CONFIG": {"owner": "platform"}}
self.write("config/deployment-setting-classification.json", json.dumps(policy))
errors = CHECKER.validate_policy(
CHECKER.load_policy(self.root / "config/deployment-setting-classification.json")
)
self.assertIn("exception FAKE_RUNTIME_CONFIG is missing reason", errors)
self.assertIn("exception FAKE_RUNTIME_CONFIG is missing expires", errors)
self.assertIn("exception FAKE_RUNTIME_CONFIG is missing allowed_sources", errors)
def test_current_tree_paths_use_git_posix_separators(self) -> None:
windows_path = PureWindowsPath(r"C:\repo\.github\workflows\deploy.yml")
windows_root = PureWindowsPath(r"C:\repo")
posix_path = PurePosixPath("/repo/.github/workflows/deploy.yml")
posix_root = PurePosixPath("/repo")
self.assertEqual(CHECKER.repository_relative_path(windows_path, windows_root), ".github/workflows/deploy.yml")
self.assertEqual(
CHECKER.repository_relative_path(windows_path, windows_root),
CHECKER.repository_relative_path(posix_path, posix_root),
)
@mock.patch.object(CHECKER.subprocess, "run")
def test_base_git_reads_decode_utf8_explicitly(self, run: mock.Mock) -> None:
run.side_effect = [
subprocess.CompletedProcess([], 0, stdout=".github/workflows/deploy.yml\n"),
subprocess.CompletedProcess([], 0, stdout="# utf-8 guard: \u2603\n"),
]
self.assertEqual(CHECKER._base_paths(self.root, "HEAD"), {".github/workflows/deploy.yml"})
self.assertEqual(
CHECKER._read_base_file(self.root, "HEAD", ".github/workflows/deploy.yml"),
"# utf-8 guard: \u2603\n",
)
self.assertEqual(len(run.call_args_list), 2)
for call in run.call_args_list:
self.assertEqual(call.kwargs["encoding"], "utf-8")
if __name__ == "__main__":
unittest.main()