mi
180 lines
7.6 KiB
Python
180 lines
7.6 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import runpy
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
VALIDATOR = ROOT / "scripts/validate-env"
|
||
|
||
|
||
def runtime_values() -> dict[str, str]:
|
||
symbols = runpy.run_path(str(VALIDATOR))
|
||
values = {key: "a" * 32 for key in symbols["REQUIRED_RUNTIME"]}
|
||
pg_tail = "?sslmode=verify-full&sslrootcert=/run/secrets/pg-ca.pem"
|
||
values.update(
|
||
{
|
||
"DATABASE_URL": f"postgresql+asyncpg://han_app:password@pg:5433/han_chat{pg_tail}",
|
||
"BITRIX_DATABASE_URL": f"postgresql://bitrix:password@pg:5433/han_chat{pg_tail}",
|
||
"SMS_DATABASE_URL": f"postgresql+asyncpg://sms:password@pg:5433/han_chat{pg_tail}",
|
||
"KEYCLOAK_DB_URL": f"jdbc:postgresql://pg:5433/han_chat{pg_tail}",
|
||
"REDIS_API_PASSWORD": "r" * 32,
|
||
"REDIS_HEALTH_PASSWORD": "h" * 32,
|
||
"REDIS_URL": f"redis://api_backend:{'r' * 32}@redis:6379/0",
|
||
"REDIS_REALTIME_URL": f"redis://api_backend:{'r' * 32}@redis:6379/1",
|
||
"BITRIX_LOCAL_APP_INTERNAL_TOKEN": "b" * 32,
|
||
"BITRIX_INTERNAL_API_TOKEN": "b" * 32,
|
||
"BITRIX_API_FORWARD_TOKEN": "f" * 32,
|
||
"BITRIX_API_INBOX_TOKEN": "f" * 32,
|
||
"KEYCLOAK_SMS_SERVICE_TOKEN": "k" * 32,
|
||
"SMS_SERVICE_TOKEN": "k" * 32,
|
||
"KEYCLOAK_OTP_MOCK_CODE": "846271",
|
||
}
|
||
)
|
||
return values
|
||
|
||
|
||
class SecretHygieneTests(unittest.TestCase):
|
||
def run_validator(
|
||
self, config: Path, *args: str, environment: dict[str, str] | None = None
|
||
) -> subprocess.CompletedProcess[str]:
|
||
return subprocess.run(
|
||
[sys.executable, str(VALIDATOR), str(config), *args],
|
||
text=True,
|
||
capture_output=True,
|
||
check=False,
|
||
env=environment,
|
||
)
|
||
|
||
def test_example_is_non_secret_config(self) -> None:
|
||
example = (ROOT / ".env.example").read_text(encoding="utf-8")
|
||
symbols = runpy.run_path(str(VALIDATOR))
|
||
for key in symbols["FORBIDDEN_CONFIG_KEYS"]:
|
||
self.assertNotRegex(example, rf"(?m)^{key}=")
|
||
self.assertIn("SECRETS_SOURCE=file", example)
|
||
|
||
def test_config_rejects_secret_key_and_credential_url(self) -> None:
|
||
example = (ROOT / ".env.example").read_text(encoding="utf-8")
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
config = Path(directory) / ".env"
|
||
config.write_text(
|
||
example
|
||
+ "\nSMS_SERVICE_TOKEN=not-for-config\n"
|
||
+ "EXTERNAL_URL=https://user:password@example.net/path\n",
|
||
encoding="utf-8",
|
||
)
|
||
result = self.run_validator(config)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("SMS_SERVICE_TOKEN", result.stderr)
|
||
self.assertIn("URL с credentials", result.stderr)
|
||
self.assertNotIn("not-for-config", result.stderr)
|
||
|
||
def test_runtime_environment_is_validated_without_value_disclosure(self) -> None:
|
||
values = runtime_values()
|
||
secret_canary = "canary-never-print-" + "x" * 20
|
||
values["CURSOR_HMAC_SECRET"] = secret_canary
|
||
environment = {**os.environ, **values}
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
config = Path(directory) / ".env"
|
||
config.write_text(
|
||
(ROOT / ".env.example").read_text(encoding="utf-8").replace(
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false",
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true",
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
result = self.run_validator(config, "--runtime-env", environment=environment)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertNotIn(secret_canary, result.stdout + result.stderr)
|
||
|
||
def test_runtime_manifest_reads_protected_files_without_value_disclosure(self) -> None:
|
||
values = runtime_values()
|
||
secret_canary = "manifest-canary-" + "z" * 20
|
||
values["CURSOR_HMAC_SECRET"] = secret_canary
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
root = Path(directory)
|
||
config = root / ".env"
|
||
config.write_text(
|
||
(ROOT / ".env.example").read_text(encoding="utf-8").replace(
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false",
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true",
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
entries = []
|
||
for index, (key, value) in enumerate(sorted(values.items())):
|
||
path = root / f"secret-{index}"
|
||
path.write_text(value + "\n", encoding="utf-8")
|
||
path.chmod(0o600)
|
||
entries.append(f"{key}={path.resolve()}")
|
||
manifest = root / "manifest"
|
||
manifest.write_text("\n".join(entries) + "\n", encoding="utf-8")
|
||
result = self.run_validator(config, "--runtime-manifest", str(manifest))
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertNotIn(secret_canary, result.stdout + result.stderr)
|
||
|
||
def test_validator_enforces_remote_safety_tls_contract(self) -> None:
|
||
example = (ROOT / ".env.example").read_text(encoding="utf-8")
|
||
cases = (
|
||
(
|
||
"MESSAGE_SAFETY_URL=https://processing.internal:8443",
|
||
"MESSAGE_SAFETY_URL=http://message-safety:8080",
|
||
"remote TLS endpoint",
|
||
),
|
||
(
|
||
"MESSAGE_SAFETY_CA_HOST_PATH=/etc/han/ca/vm2-internal-ca.pem",
|
||
"MESSAGE_SAFETY_CA_HOST_PATH=",
|
||
"MESSAGE_SAFETY_CA_HOST_PATH",
|
||
),
|
||
(
|
||
"MESSAGE_SAFETY_EXTRA_HOST=processing.internal=192.168.0.4",
|
||
"MESSAGE_SAFETY_EXTRA_HOST=processing.internal=8.8.8.8",
|
||
"MESSAGE_SAFETY_EXTRA_HOST",
|
||
),
|
||
(
|
||
"MESSAGE_SAFETY_API_PREFIX=/internal/safety/v2",
|
||
"MESSAGE_SAFETY_API_PREFIX=/internal/safety/v1",
|
||
"MESSAGE_SAFETY_API_PREFIX",
|
||
),
|
||
)
|
||
for original, replacement, expected_error in cases:
|
||
with self.subTest(replacement=replacement), tempfile.TemporaryDirectory() as directory:
|
||
config = Path(directory) / ".env"
|
||
config.write_text(
|
||
example.replace(original, replacement),
|
||
encoding="utf-8",
|
||
)
|
||
result = self.run_validator(config)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn(expected_error, result.stderr)
|
||
|
||
def test_validator_accepts_configured_private_safety_hostname(self) -> None:
|
||
example = (ROOT / ".env.example").read_text(encoding="utf-8")
|
||
configured = (
|
||
example.replace(
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false",
|
||
"KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true",
|
||
)
|
||
.replace(
|
||
"MESSAGE_SAFETY_URL=https://processing.internal:8443",
|
||
"MESSAGE_SAFETY_URL=https://safety.vm2.corp.internal:8443",
|
||
)
|
||
.replace(
|
||
"MESSAGE_SAFETY_EXTRA_HOST=processing.internal=192.168.0.4",
|
||
"MESSAGE_SAFETY_EXTRA_HOST=safety.vm2.corp.internal=192.168.0.4",
|
||
)
|
||
)
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
config = Path(directory) / ".env"
|
||
config.write_text(configured, encoding="utf-8")
|
||
result = self.run_validator(config)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|