Перенесены секреты из .env в SM
This commit is contained in:
@@ -0,0 +1,126 @@
|
||||
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}",
|
||||
"BITRIX_SYNC_DATABASE_URL": f"postgresql://sync:password@pg:5433/han_chat{pg_tail}",
|
||||
"MESSAGE_SAFETY_DATABASE_URL": f"postgresql://safety: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_SAFETY_PASSWORD": "s" * 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",
|
||||
"MESSAGE_SAFETY_REDIS_URL": f"redis://message_safety:{'s' * 32}@redis:6379/2",
|
||||
"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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user