alex
51d93d9dc6
- 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
136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
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)}"
|
|
)
|