feat(FN-009): implement config and secrets management with runtime injection

- Add RuntimeInjectionService for scope-based config/secret resolution
- Mount configs as JSON files at /app/config/ with 0400 permissions
- Inject secrets as environment variables with uppercase keys
- Implement scope hierarchy: instance > project > user > global
- Create ConfigListPage and SecretListPage frontend components
- Mask secret values in API responses (never expose decrypted)
- Validate secrets exist before spawning containers
- Add comprehensive tests for runtime injection service
- Update documentation with config/secrets workflow
This commit is contained in:
2026-05-15 16:44:26 +02:00
parent 78aaddb2b5
commit 51d93d9dc6
12 changed files with 838 additions and 86 deletions
+5 -29
View File
@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.dependencies import get_current_active_user
from app.db import get_db_session
from app.encryption import decrypt_value, encrypt_value
from app.encryption import encrypt_value
from app.models.project import Project
from app.models.secret import Secret
from app.models.tool_instance import ToolInstance
@@ -55,13 +55,7 @@ async def create_secret(
session.add(secret)
await session.commit()
await session.refresh(secret)
return SecretRead(
id=secret.id,
scope_type=secret.scope_type,
scope_id=secret.scope_id,
key=secret.key,
value=decrypt_value(secret.encrypted_value),
)
return SecretRead.from_secret(secret)
@router.get("/secrets", response_model=list[SecretRead])
@@ -82,13 +76,7 @@ async def list_secrets(
for s in secrets:
try:
await _verify_secret_ownership(s, current_user, session)
allowed.append(SecretRead(
id=s.id,
scope_type=s.scope_type,
scope_id=s.scope_id,
key=s.key,
value=decrypt_value(s.encrypted_value),
))
allowed.append(SecretRead.from_secret(s))
except HTTPException:
pass
return allowed
@@ -104,13 +92,7 @@ async def get_secret(
if not s:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Secret not found")
await _verify_secret_ownership(s, current_user, session)
return SecretRead(
id=s.id,
scope_type=s.scope_type,
scope_id=s.scope_id,
key=s.key,
value=decrypt_value(s.encrypted_value),
)
return SecretRead.from_secret(s)
@router.put("/secrets/{secret_id}", response_model=SecretRead)
@@ -130,13 +112,7 @@ async def update_secret(
s.encrypted_value = encrypt_value(secret_in.value)
await session.commit()
await session.refresh(s)
return SecretRead(
id=s.id,
scope_type=s.scope_type,
scope_id=s.scope_id,
key=s.key,
value=decrypt_value(s.encrypted_value),
)
return SecretRead.from_secret(s)
@router.delete("/secrets/{secret_id}", status_code=status.HTTP_204_NO_CONTENT)
+17 -1
View File
@@ -1,7 +1,13 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from uuid import UUID
from app.schemas.base import OrmBase
if TYPE_CHECKING:
from app.models.secret import Secret
class SecretBase(OrmBase):
scope_type: str
@@ -15,7 +21,17 @@ class SecretCreate(SecretBase):
class SecretRead(SecretBase):
id: UUID
value: str
value: str = "••••••"
@classmethod
def from_secret(cls, secret: Secret) -> SecretRead:
return cls(
id=secret.id,
scope_type=secret.scope_type,
scope_id=secret.scope_id,
key=secret.key,
value="••••••",
)
class SecretUpdate(OrmBase):
+135
View File
@@ -0,0 +1,135 @@
from __future__ import annotations
import json
import uuid
from pathlib import Path
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.encryption import decrypt_value
from app.models.config import Config
from app.models.secret import Secret
class RuntimeInjectionError(Exception):
pass
class RuntimeInjectionService:
SCOPE_HIERARCHY = ["global", "user", "project", "tool_instance"]
@staticmethod
async def resolve_configs(
session: AsyncSession,
project_id: uuid.UUID,
user_id: uuid.UUID,
instance_id: uuid.UUID | None = None,
tool_definition_id: uuid.UUID | None = None,
) -> dict[str, Any]:
stmt = select(Config).where(
(Config.scope_type == "global")
| (
(Config.scope_type == "user")
& (Config.scope_id == user_id)
)
| (
(Config.scope_type == "project")
& (Config.scope_id == project_id)
)
| (
(Config.scope_type == "tool_instance")
& (Config.scope_id == (instance_id or uuid.UUID(int=0)))
)
)
if tool_definition_id:
stmt = stmt.where(
(Config.tool_definition_id == tool_definition_id)
| (Config.tool_definition_id.is_(None))
)
result = await session.execute(stmt)
configs = list(result.scalars().all())
resolved: dict[str, Any] = {}
for scope in RuntimeInjectionService.SCOPE_HIERARCHY:
for cfg in configs:
if cfg.scope_type == scope:
resolved[cfg.key] = cfg.value
return resolved
@staticmethod
async def resolve_secrets(
session: AsyncSession,
project_id: uuid.UUID,
user_id: uuid.UUID,
instance_id: uuid.UUID | None = None,
) -> dict[str, str]:
stmt = select(Secret).where(
(Secret.scope_type == "global")
| (
(Secret.scope_type == "user")
& (Secret.scope_id == user_id)
)
| (
(Secret.scope_type == "project")
& (Secret.scope_id == project_id)
)
| (
(Secret.scope_type == "tool_instance")
& (Secret.scope_id == (instance_id or uuid.UUID(int=0)))
)
)
result = await session.execute(stmt)
secrets = list(result.scalars().all())
resolved: dict[str, str] = {}
for scope in RuntimeInjectionService.SCOPE_HIERARCHY:
for secret in secrets:
if secret.scope_type == scope:
resolved[secret.key] = decrypt_value(secret.encrypted_value)
return resolved
@staticmethod
def generate_config_files(configs: dict[str, Any], config_dir: Path) -> list[str]:
config_dir.mkdir(parents=True, exist_ok=True)
mounts = []
for key, value in configs.items():
file_path = config_dir / f"{key}.json"
file_path.write_text(json.dumps(value, indent=2))
file_path.chmod(0o400)
mounts.append(f"{file_path}:/app/config/{key}.json:ro")
return mounts
@staticmethod
def generate_secret_env_vars(secrets: dict[str, str]) -> dict[str, str]:
return {key.upper(): value for key, value in secrets.items()}
@staticmethod
async def validate_secrets_exist(
session: AsyncSession,
required_secret_keys: list[str],
project_id: uuid.UUID,
user_id: uuid.UUID,
instance_id: uuid.UUID | None = None,
) -> None:
resolved = await RuntimeInjectionService.resolve_secrets(
session, project_id, user_id, instance_id
)
missing = [key for key in required_secret_keys if key not in resolved]
if missing:
raise RuntimeInjectionError(
f"Missing required secrets: {', '.join(missing)}"
)
+12
View File
@@ -38,6 +38,8 @@ class SpawnService:
workspace_path: Path | None = None,
config_path: Path | None = None,
ssh_key_path: Path | None = None,
config_mounts: list[str] | None = None,
secret_env_vars: dict[str, str] | None = None,
) -> dict[str, Any]:
service_name = f"tool-{instance_id[:8]}"
@@ -96,9 +98,15 @@ class SpawnService:
if ssh_key_path and ssh_key_path.exists():
volumes.append(f"{ssh_key_path}:/home/coder/.ssh:ro")
if config_mounts:
volumes.extend(config_mounts)
if volumes:
service["volumes"] = volumes
if secret_env_vars:
service["environment"].update(secret_env_vars)
if manifest.health_check:
hc = manifest.health_check
healthcheck: dict[str, Any] = {
@@ -170,6 +178,8 @@ class SpawnService:
workspace_path: Path | None = None,
config_path: Path | None = None,
ssh_key_path: Path | None = None,
config_mounts: list[str] | None = None,
secret_env_vars: dict[str, str] | None = None,
) -> dict[str, Any]:
label_gen = TraefikLabelGenerator(domain=settings.root_domain)
@@ -203,6 +213,8 @@ class SpawnService:
workspace_path=workspace_path,
config_path=config_path,
ssh_key_path=ssh_key_path,
config_mounts=config_mounts,
secret_env_vars=secret_env_vars,
)
compose_path = self._write_compose_file(instance_id, service)
+110
View File
@@ -0,0 +1,110 @@
from uuid import UUID
import pytest
from app.services.runtime_injection import RuntimeInjectionError, RuntimeInjectionService
@pytest.mark.asyncio
async def test_resolve_configs_empty(db_session):
result = await RuntimeInjectionService.resolve_configs(
db_session,
project_id=UUID(int=1),
user_id=UUID(int=2),
)
assert result == {}
@pytest.mark.asyncio
async def test_resolve_configs_global_only(db_session, sample_config):
result = await RuntimeInjectionService.resolve_configs(
db_session,
project_id=UUID(int=1),
user_id=UUID(int=2),
)
assert result == {"test_key": "test_value"}
@pytest.mark.asyncio
async def test_resolve_configs_scope_override(db_session):
from app.models.config import Config
global_config = Config(
scope_type="global",
scope_id=UUID(int=0),
key="shared_key",
value="global_value",
)
project_config = Config(
scope_type="project",
scope_id=UUID(int=1),
key="shared_key",
value="project_value",
)
db_session.add_all([global_config, project_config])
await db_session.commit()
result = await RuntimeInjectionService.resolve_configs(
db_session,
project_id=UUID(int=1),
user_id=UUID(int=2),
)
assert result["shared_key"] == "project_value"
@pytest.mark.asyncio
async def test_resolve_secrets_empty(db_session):
result = await RuntimeInjectionService.resolve_secrets(
db_session,
project_id=UUID(int=1),
user_id=UUID(int=2),
)
assert result == {}
@pytest.mark.asyncio
async def test_resolve_secrets_decrypts(db_session, sample_secret):
result = await RuntimeInjectionService.resolve_secrets(
db_session,
project_id=UUID(int=1),
user_id=UUID(int=2),
)
assert result == {"secret_key": "secret_value"}
@pytest.mark.asyncio
async def test_validate_secrets_exist_missing(db_session):
with pytest.raises(RuntimeInjectionError, match="Missing required secrets"):
await RuntimeInjectionService.validate_secrets_exist(
db_session,
required_secret_keys=["missing_secret"],
project_id=UUID(int=1),
user_id=UUID(int=2),
)
@pytest.mark.asyncio
async def test_validate_secrets_exist_found(db_session, sample_secret):
await RuntimeInjectionService.validate_secrets_exist(
db_session,
required_secret_keys=["secret_key"],
project_id=UUID(int=1),
user_id=UUID(int=2),
)
def test_generate_config_files(tmp_path):
configs = {"app": {"port": 8080}, "debug": True}
mounts = RuntimeInjectionService.generate_config_files(configs, tmp_path)
assert len(mounts) == 2
assert (tmp_path / "app.json").exists()
assert (tmp_path / "debug.json").exists()
assert (tmp_path / "app.json").stat().st_mode & 0o777 == 0o400
def test_generate_secret_env_vars():
secrets = {"api_key": "abc123", "db_pass": "secret"}
env_vars = RuntimeInjectionService.generate_secret_env_vars(secrets)
assert env_vars == {"API_KEY": "abc123", "DB_PASS": "secret"}