mi
472 lines
17 KiB
Python
472 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LOADER_PATH = ROOT / "deployment" / "secrets" / "secrets_loader.py"
|
|
LAUNCHER_PATH = ROOT / "deployment" / "secrets" / "han-secrets"
|
|
SPEC = importlib.util.spec_from_file_location("han_secrets_loader", LOADER_PATH)
|
|
assert SPEC and SPEC.loader
|
|
loader = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = loader
|
|
SPEC.loader.exec_module(loader)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(
|
|
self, status: int, body: bytes, headers: dict[str, str] | None = None
|
|
) -> None:
|
|
self.status = status
|
|
self._body = body
|
|
self.headers = headers or {}
|
|
|
|
def __enter__(self) -> "FakeResponse":
|
|
return self
|
|
|
|
def __exit__(self, *args: Any) -> None:
|
|
return None
|
|
|
|
def read(self, amount: int) -> bytes:
|
|
return self._body[:amount]
|
|
|
|
|
|
class FakeOpener:
|
|
def __init__(self, outcomes: list[Any]) -> None:
|
|
self.outcomes = outcomes
|
|
self.requests: list[Any] = []
|
|
|
|
def open(self, request: Any, timeout: float) -> FakeResponse:
|
|
self.requests.append((request, timeout))
|
|
outcome = self.outcomes.pop(0)
|
|
if isinstance(outcome, BaseException):
|
|
raise outcome
|
|
return outcome
|
|
|
|
|
|
def identity_response(*, project_scoped: bool = True) -> FakeResponse:
|
|
token: dict[str, Any] = {
|
|
"catalog": [
|
|
{
|
|
"type": "secrets-manager",
|
|
"endpoints": [
|
|
{
|
|
"region": "ru-test",
|
|
"interface": "public",
|
|
"url": "https://secrets.example",
|
|
}
|
|
],
|
|
}
|
|
]
|
|
}
|
|
if project_scoped:
|
|
token["project"] = {"id": "project-id"}
|
|
return FakeResponse(
|
|
201,
|
|
json.dumps({"token": token}).encode(),
|
|
{"X-Subject-Token": "iam-token-do-not-log"},
|
|
)
|
|
|
|
|
|
def secret_response(value: bytes) -> FakeResponse:
|
|
return FakeResponse(
|
|
200, json.dumps({"value": base64.b64encode(value).decode()}).encode()
|
|
)
|
|
|
|
|
|
def selectel_config(runtime_dir: Path) -> dict[str, Any]:
|
|
return {
|
|
"version": 1,
|
|
"mode": "selectel",
|
|
"runtime_dir": str(runtime_dir),
|
|
"http": {"timeout_seconds": 2, "retries": 1, "max_response_bytes": 4096},
|
|
"selectel": {
|
|
"account_id": "123456",
|
|
"username": "reader",
|
|
"project_name": "production",
|
|
"region": "ru-test",
|
|
"password_file": "selectel-password",
|
|
},
|
|
"secrets": {
|
|
"DATABASE_URL": {
|
|
"remote": "han/database-url",
|
|
"consumers": ["api", "migration"],
|
|
"max_bytes": 512,
|
|
},
|
|
"SHARED_TOKEN": {
|
|
"remote": "han/shared-token",
|
|
"consumers": ["api"],
|
|
"max_bytes": 128,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def write_json(path: Path, value: dict[str, Any]) -> None:
|
|
path.write_text(json.dumps(value), encoding="utf-8")
|
|
|
|
|
|
def make_private(path: Path) -> None:
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
class SecretsLoaderTests(unittest.TestCase):
|
|
def test_decodes_current_secret_from_nested_selectel_version(self) -> None:
|
|
encoded = base64.b64encode(b"current-secret").decode()
|
|
|
|
value = loader.decode_secret(
|
|
{"name": "DATABASE_URL", "version": {"version_id": 1, "value": encoded}},
|
|
"DATABASE_URL",
|
|
1024,
|
|
)
|
|
|
|
self.assertEqual(value, b"current-secret")
|
|
|
|
def test_empty_literal_never_requires_provider_or_fallback_value(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
fallback = root / "fallback.env"
|
|
fallback.write_text("REQUIRED_TOKEN=required-value\n", encoding="utf-8")
|
|
make_private(fallback)
|
|
config = {
|
|
"version": 1,
|
|
"mode": "file",
|
|
"runtime_dir": str(root / "run"),
|
|
"file": {"path": str(fallback)},
|
|
"secrets": {
|
|
"REQUIRED_TOKEN": {
|
|
"remote": "provider-token",
|
|
"consumers": ["service"],
|
|
},
|
|
"OPTIONAL_HEADER": {
|
|
"literal": "",
|
|
"consumers": ["service"],
|
|
},
|
|
},
|
|
}
|
|
config_path = root / "file.json"
|
|
write_json(config_path, config)
|
|
|
|
loader.run(config_path, environ={})
|
|
|
|
self.assertEqual((root / "run" / "OPTIONAL_HEADER").read_bytes(), b"")
|
|
self.assertIn(
|
|
'OPTIONAL_HEADER=""',
|
|
(root / "run" / "service.env").read_text(encoding="utf-8"),
|
|
)
|
|
|
|
def test_launcher_uses_public_source_switch_without_exporting_values(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
runtime = root / "runtime"
|
|
fallback = root / "fallback.env"
|
|
fallback.write_text("TEST_SECRET=canary-secret-value\n", encoding="utf-8")
|
|
make_private(fallback)
|
|
public = root / ".env"
|
|
public.write_text(
|
|
"SECRETS_SOURCE=file\nAPP_ENV=production\n", encoding="utf-8"
|
|
)
|
|
loader_config = root / "production.file.json"
|
|
write_json(
|
|
loader_config,
|
|
{
|
|
"version": 1,
|
|
"mode": "file",
|
|
"runtime_dir": str(runtime),
|
|
"file": {"path": str(fallback)},
|
|
"secrets": {
|
|
"TEST_SECRET": {
|
|
"remote": "unused-in-file-mode",
|
|
"consumers": ["test-service"],
|
|
}
|
|
},
|
|
},
|
|
)
|
|
probe = (
|
|
"import json,os,pathlib;"
|
|
"p=os.environ['TEST_SECRET_FILE'];"
|
|
"print(json.dumps({'raw':os.environ.get('TEST_SECRET'),"
|
|
"'value':pathlib.Path(p).read_text()}))"
|
|
)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(LAUNCHER_PATH),
|
|
"run",
|
|
"--config",
|
|
str(public),
|
|
"--loader-config",
|
|
str(loader_config),
|
|
"--",
|
|
sys.executable,
|
|
"-c",
|
|
probe,
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertIsNone(payload["raw"])
|
|
self.assertEqual(payload["value"], "canary-secret-value")
|
|
self.assertEqual(
|
|
(runtime / "manifest").read_text(encoding="utf-8").strip(),
|
|
f"TEST_SECRET={(runtime / 'TEST_SECRET').resolve()}",
|
|
)
|
|
|
|
def test_selectel_flow_uses_project_scope_catalog_and_per_service_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
credentials = root / "credentials"
|
|
credentials.mkdir()
|
|
(credentials / "selectel-password").write_text(
|
|
"service-user-password\n", encoding="utf-8"
|
|
)
|
|
make_private(credentials / "selectel-password")
|
|
runtime = root / "run"
|
|
config_path = root / "config.json"
|
|
write_json(config_path, selectel_config(runtime))
|
|
opener = FakeOpener(
|
|
[
|
|
identity_response(),
|
|
secret_response(b"postgresql://user:password@db/app"),
|
|
secret_response(b'token with spaces and "quotes"'),
|
|
]
|
|
)
|
|
|
|
def factory(**kwargs: Any) -> Any:
|
|
return loader.HTTPClient(
|
|
**kwargs, opener=opener, sleeper=lambda _: None, jitter=lambda: 0
|
|
)
|
|
|
|
consumers = loader.run(
|
|
config_path,
|
|
environ={"CREDENTIALS_DIRECTORY": str(credentials)},
|
|
client_factory=factory,
|
|
)
|
|
|
|
self.assertEqual(consumers, ["api", "migration"])
|
|
auth_request = opener.requests[0][0]
|
|
auth_payload = json.loads(auth_request.data)
|
|
self.assertEqual(
|
|
auth_payload["auth"]["scope"]["project"]["name"], "production"
|
|
)
|
|
self.assertEqual(
|
|
auth_payload["auth"]["identity"]["password"]["user"]["domain"]["name"],
|
|
"123456",
|
|
)
|
|
self.assertEqual(
|
|
opener.requests[1][0].get_header("X-auth-token"),
|
|
"iam-token-do-not-log",
|
|
)
|
|
self.assertEqual(
|
|
opener.requests[1][0].full_url,
|
|
"https://secrets.example/v1/han%2Fdatabase-url",
|
|
)
|
|
api_text = (runtime / "api.env").read_text(encoding="utf-8")
|
|
migration_text = (runtime / "migration.env").read_text(encoding="utf-8")
|
|
self.assertIn('DATABASE_URL="postgresql://user:password@db/app"', api_text)
|
|
self.assertIn(
|
|
'SHARED_TOKEN="token with spaces and \\"quotes\\""', api_text
|
|
)
|
|
self.assertNotIn("SHARED_TOKEN", migration_text)
|
|
if os.name != "nt":
|
|
self.assertEqual(stat.S_IMODE((runtime / "api.env").stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(runtime.stat().st_mode), 0o700)
|
|
|
|
def test_retry_uses_jitter_and_then_succeeds(self) -> None:
|
|
opener = FakeOpener(
|
|
[urllib.error.URLError("temporary"), FakeResponse(200, b"{}")]
|
|
)
|
|
sleeps: list[float] = []
|
|
client = loader.HTTPClient(
|
|
timeout=1,
|
|
retries=1,
|
|
max_response_bytes=1024,
|
|
opener=opener,
|
|
sleeper=sleeps.append,
|
|
jitter=lambda: 0.25,
|
|
)
|
|
result = client.request(
|
|
"GET", "https://provider.example/v1/key", expected=frozenset({200})
|
|
)
|
|
self.assertEqual(result.status, 200)
|
|
self.assertEqual(sleeps, [0.1875])
|
|
|
|
def test_paired_canonical_names_fetch_provider_version_once(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
credentials = root / "credentials"
|
|
credentials.mkdir()
|
|
credential = credentials / "selectel-password"
|
|
credential.write_text("password", encoding="utf-8")
|
|
make_private(credential)
|
|
config = selectel_config(root / "run")
|
|
config["secrets"]["SHARED_TOKEN"]["remote"] = "han/database-url"
|
|
config_path = root / "config.json"
|
|
write_json(config_path, config)
|
|
opener = FakeOpener([identity_response(), secret_response(b"same-value")])
|
|
|
|
def factory(**kwargs: Any) -> Any:
|
|
return loader.HTTPClient(**kwargs, opener=opener)
|
|
|
|
loader.run(
|
|
config_path,
|
|
environ={"CREDENTIALS_DIRECTORY": str(credentials)},
|
|
client_factory=factory,
|
|
)
|
|
self.assertEqual(len(opener.requests), 2)
|
|
output = (root / "run" / "api.env").read_text(encoding="utf-8")
|
|
self.assertIn('DATABASE_URL="same-value"', output)
|
|
self.assertIn('SHARED_TOKEN="same-value"', output)
|
|
|
|
def test_http_error_is_redacted_and_body_is_not_read(self) -> None:
|
|
leaked = b"postgresql://admin:secret@db/app iam-token response-body"
|
|
|
|
class ExplodingBody(io.BytesIO):
|
|
def read(self, *args: Any, **kwargs: Any) -> bytes:
|
|
raise AssertionError("HTTP error body must not be read")
|
|
|
|
error = urllib.error.HTTPError(
|
|
"https://provider.example/v1/key",
|
|
403,
|
|
"body contains a secret",
|
|
{},
|
|
ExplodingBody(leaked),
|
|
)
|
|
client = loader.HTTPClient(
|
|
timeout=1,
|
|
retries=0,
|
|
max_response_bytes=1024,
|
|
opener=FakeOpener([error]),
|
|
)
|
|
with self.assertRaises(loader.LoaderError) as caught:
|
|
client.request(
|
|
"GET",
|
|
"https://provider.example/v1/key",
|
|
headers={"X-Auth-Token": "iam-token"},
|
|
expected=frozenset({200}),
|
|
)
|
|
message = str(caught.exception)
|
|
self.assertEqual(message, "provider request failed with HTTP 403")
|
|
for forbidden in ("secret", "iam-token", "response-body", "postgresql://"):
|
|
self.assertNotIn(forbidden, message)
|
|
|
|
def test_unscoped_token_fails_without_replacing_existing_output(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
credentials = root / "credentials"
|
|
credentials.mkdir()
|
|
(credentials / "selectel-password").write_text("password", encoding="utf-8")
|
|
make_private(credentials / "selectel-password")
|
|
runtime = root / "run"
|
|
runtime.mkdir()
|
|
existing = runtime / "api.env"
|
|
existing.write_text('DATABASE_URL="old-value"\n', encoding="utf-8")
|
|
config_path = root / "config.json"
|
|
write_json(config_path, selectel_config(runtime))
|
|
opener = FakeOpener([identity_response(project_scoped=False)])
|
|
|
|
def factory(**kwargs: Any) -> Any:
|
|
return loader.HTTPClient(**kwargs, opener=opener)
|
|
|
|
with self.assertRaisesRegex(loader.LoaderError, "not project-scoped"):
|
|
loader.run(
|
|
config_path,
|
|
environ={"CREDENTIALS_DIRECTORY": str(credentials)},
|
|
client_factory=factory,
|
|
)
|
|
self.assertEqual(
|
|
existing.read_text(encoding="utf-8"),
|
|
'DATABASE_URL="old-value"\n',
|
|
)
|
|
|
|
def test_invalid_base64_and_size_limit_fail_closed(self) -> None:
|
|
with self.assertRaisesRegex(loader.LoaderError, "invalid base64"):
|
|
loader.decode_secret({"value": "not-base64!"}, "TOKEN", 128)
|
|
with self.assertRaisesRegex(loader.LoaderError, "configured limit"):
|
|
loader.decode_secret(
|
|
{"value": base64.b64encode(b"too-long").decode()}, "TOKEN", 3
|
|
)
|
|
with self.assertRaisesRegex(loader.LoaderError, "dotenv"):
|
|
loader.decode_secret(
|
|
{"value": base64.b64encode(b"line1\nline2").decode()}, "TOKEN", 128
|
|
)
|
|
|
|
def test_file_mode_is_explicit_strict_and_narrow(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "fallback.env"
|
|
source.write_text(
|
|
"# exact recovery set\n"
|
|
'DATABASE_URL="postgresql://user:pass@db/app"\n'
|
|
"SHARED_TOKEN='literal value'\n",
|
|
encoding="utf-8",
|
|
)
|
|
make_private(source)
|
|
runtime = root / "run"
|
|
config = selectel_config(runtime)
|
|
config["mode"] = "file"
|
|
config.pop("selectel")
|
|
config.pop("http")
|
|
config["file"] = {"path": str(source)}
|
|
config_path = root / "config.json"
|
|
write_json(config_path, config)
|
|
loader.run(config_path, environ={})
|
|
output = (runtime / "api.env").read_text(encoding="utf-8")
|
|
self.assertIn("literal value", output)
|
|
self.assertNotIn("selectel", output)
|
|
|
|
source.write_text(
|
|
"DATABASE_URL=ok\nSHARED_TOKEN=ok\nUNDECLARED=leak\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(loader.LoaderError, "undeclared key"):
|
|
loader.run(config_path, environ={})
|
|
|
|
def test_selectel_failure_never_falls_back_to_file_section(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
config = selectel_config(root / "run")
|
|
config["file"] = {"path": str(root / "fallback.env")}
|
|
config_path = root / "config.json"
|
|
write_json(config_path, config)
|
|
with self.assertRaisesRegex(loader.LoaderError, "forbidden"):
|
|
loader.run(config_path, environ={})
|
|
|
|
def test_relative_credential_requires_systemd_directory(self) -> None:
|
|
selectel = {"password_file": "credential"}
|
|
with self.assertRaisesRegex(loader.LoaderError, "CREDENTIALS_DIRECTORY"):
|
|
loader.credential_value(selectel, {})
|
|
|
|
def test_response_content_length_limit_is_enforced(self) -> None:
|
|
client = loader.HTTPClient(
|
|
timeout=1,
|
|
retries=0,
|
|
max_response_bytes=10,
|
|
opener=FakeOpener(
|
|
[FakeResponse(200, b"{}", {"Content-Length": "100"})]
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(loader.LoaderError, "exceeds"):
|
|
client.request(
|
|
"GET", "https://provider.example/v1/key", expected=frozenset({200})
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|