190 lines
7.0 KiB
Python
190 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
from typing import Any, Dict, Iterable, Optional
|
|
|
|
DEFAULTS: Dict[str, Any] = {
|
|
"container_prefix": "fis-",
|
|
"environment": {},
|
|
"volumes": {},
|
|
"mem_limit": None,
|
|
"memswap_limit": None,
|
|
"privileged": False,
|
|
"tmpfs": {},
|
|
"dns": [],
|
|
"extra_hosts": {},
|
|
}
|
|
|
|
|
|
def merge_dict(base: Optional[Dict[str, Any]], override: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
|
result = dict(base or {})
|
|
result.update(override or {})
|
|
return result
|
|
|
|
|
|
def resolve_placeholders(value: Any, variables: Dict[str, Any]) -> Any:
|
|
if isinstance(value, str):
|
|
return value.format(**variables)
|
|
if isinstance(value, dict):
|
|
return {
|
|
resolve_placeholders(key, variables): resolve_placeholders(item, variables)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [resolve_placeholders(item, variables) for item in value]
|
|
return value
|
|
|
|
|
|
def merge_defaults(config: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any]:
|
|
defaults = merge_dict(DEFAULTS, config.get("defaults"))
|
|
merged = dict(user)
|
|
|
|
merged["id"] = str(user["id"])
|
|
merged["port"] = int(user["port"])
|
|
variables = {"id": merged["id"], "port": merged["port"]}
|
|
|
|
merged["environment"] = merge_dict(
|
|
resolve_placeholders(defaults.get("environment", {}), variables),
|
|
resolve_placeholders(user.get("environment"), variables),
|
|
)
|
|
merged["environment"].setdefault("CUSTOM_USER", merged["id"])
|
|
merged["volumes"] = merge_dict(
|
|
resolve_placeholders(defaults.get("volumes", {}), variables),
|
|
resolve_placeholders(user.get("volumes"), variables),
|
|
)
|
|
merged["extra_hosts"] = merge_dict(
|
|
resolve_placeholders(defaults.get("extra_hosts", {}), variables),
|
|
resolve_placeholders(user.get("extra_hosts"), variables),
|
|
)
|
|
merged["tmpfs"] = merge_dict(
|
|
resolve_placeholders(defaults.get("tmpfs", {}), variables),
|
|
resolve_placeholders(user.get("tmpfs"), variables),
|
|
)
|
|
merged["dns"] = resolve_placeholders(user.get("dns", defaults.get("dns", [])), variables)
|
|
merged["mem_limit"] = user.get("mem_limit", defaults.get("mem_limit"))
|
|
merged["memswap_limit"] = user.get("memswap_limit", defaults.get("memswap_limit"))
|
|
merged["privileged"] = user.get("privileged", defaults.get("privileged", False))
|
|
|
|
if not merged.get("container_name"):
|
|
prefix = defaults.get("container_prefix", "fis-")
|
|
merged["container_name"] = f"{prefix}{merged['id']}"
|
|
|
|
if not merged["volumes"]:
|
|
merged["volumes"] = {f"v-conf-{merged['id']}": "/config"}
|
|
|
|
return merged
|
|
|
|
|
|
def validate_users(users: Iterable[Dict[str, Any]]) -> None:
|
|
seen_ids = set()
|
|
seen_ports = set()
|
|
seen_names = set()
|
|
|
|
for user in users:
|
|
uid = user["id"]
|
|
port = user["port"]
|
|
name = user["container_name"]
|
|
mount_targets = list((user.get("volumes") or {}).values())
|
|
|
|
if uid in seen_ids:
|
|
raise ValueError(f"Duplicate user id: {uid}")
|
|
if port in seen_ports:
|
|
raise ValueError(f"Duplicate port: {port}")
|
|
if name in seen_names:
|
|
raise ValueError(f"Duplicate container_name: {name}")
|
|
if not user["volumes"]:
|
|
raise ValueError(f"User {uid} must define at least one volume")
|
|
if len(mount_targets) != len(set(mount_targets)):
|
|
raise ValueError(f"User {uid} has duplicate volume mount targets")
|
|
if "PASSWORD" not in user["environment"] or not user["environment"]["PASSWORD"]:
|
|
raise ValueError(f"User {uid} must define environment.PASSWORD")
|
|
|
|
seen_ids.add(uid)
|
|
seen_ports.add(port)
|
|
seen_names.add(name)
|
|
|
|
|
|
def normalize_config(raw: Dict[str, Any]) -> Dict[str, Any]:
|
|
raw_users = raw.get("users", [])
|
|
if not raw_users:
|
|
raise ValueError("Configuration must contain at least one user")
|
|
|
|
users = [merge_defaults(raw, user) for user in raw_users]
|
|
validate_users(users)
|
|
|
|
return {"defaults": merge_dict(DEFAULTS, raw.get("defaults")), "users": users}
|
|
|
|
|
|
def compact_config(raw: Dict[str, Any]) -> Dict[str, Any]:
|
|
normalized = normalize_config(raw)
|
|
defaults = merge_dict(DEFAULTS, raw.get("defaults"))
|
|
compact_users = []
|
|
|
|
for user in normalized["users"]:
|
|
user_id = user["id"]
|
|
variables = {"id": user_id, "port": user["port"]}
|
|
resolved_default_env = resolve_placeholders(defaults.get("environment", {}), variables)
|
|
resolved_default_volumes = resolve_placeholders(defaults.get("volumes", {}), variables)
|
|
resolved_default_extra_hosts = resolve_placeholders(defaults.get("extra_hosts", {}), variables)
|
|
resolved_default_tmpfs = resolve_placeholders(defaults.get("tmpfs", {}), variables)
|
|
resolved_default_dns = resolve_placeholders(defaults.get("dns", []), variables)
|
|
default_container_name = f"{defaults.get('container_prefix', 'fis-')}{user_id}"
|
|
|
|
compact_user: Dict[str, Any] = {
|
|
"id": user_id,
|
|
"port": user["port"],
|
|
}
|
|
|
|
if user.get("container_name") and user["container_name"] != default_container_name:
|
|
compact_user["container_name"] = user["container_name"]
|
|
|
|
environment = {
|
|
key: value
|
|
for key, value in (user.get("environment") or {}).items()
|
|
if resolved_default_env.get(key) != value and not (key == "CUSTOM_USER" and value == user_id)
|
|
}
|
|
if environment:
|
|
compact_user["environment"] = environment
|
|
|
|
volumes = {
|
|
key: value
|
|
for key, value in (user.get("volumes") or {}).items()
|
|
if resolved_default_volumes.get(key) != value
|
|
}
|
|
if volumes:
|
|
compact_user["volumes"] = volumes
|
|
|
|
extra_hosts = {
|
|
key: value
|
|
for key, value in (user.get("extra_hosts") or {}).items()
|
|
if resolved_default_extra_hosts.get(key) != value
|
|
}
|
|
if extra_hosts:
|
|
compact_user["extra_hosts"] = extra_hosts
|
|
|
|
tmpfs = {
|
|
key: value
|
|
for key, value in (user.get("tmpfs") or {}).items()
|
|
if resolved_default_tmpfs.get(key) != value
|
|
}
|
|
if tmpfs:
|
|
compact_user["tmpfs"] = tmpfs
|
|
|
|
if user.get("dns") != resolved_default_dns and user.get("dns") is not None:
|
|
compact_user["dns"] = user["dns"]
|
|
if user.get("mem_limit") != defaults.get("mem_limit"):
|
|
compact_user["mem_limit"] = user.get("mem_limit")
|
|
if user.get("memswap_limit") != defaults.get("memswap_limit"):
|
|
compact_user["memswap_limit"] = user.get("memswap_limit")
|
|
if user.get("privileged") != defaults.get("privileged", False):
|
|
compact_user["privileged"] = user.get("privileged", False)
|
|
|
|
compact_users.append(compact_user)
|
|
|
|
return {"defaults": defaults, "users": compact_users}
|
|
|
|
|
|
def load_config(path: str) -> Dict[str, Any]:
|
|
with open(path, encoding="utf-8") as handle:
|
|
raw = json.load(handle)
|
|
return normalize_config(raw)
|