feat(services): declarative config_schema on BaseService + SSH/HTTP/HTTPS descriptors

ServiceConfigField dataclass + BaseService.validate_cfg coerce/drop submitted
service_cfg dicts against per-service typed schemas. SSH/HTTP/HTTPS now declare
the keys they already read in compose_fragment, so the upcoming Inspector form
has metadata to render from instead of hardcoded inputs per service.
This commit is contained in:
2026-04-29 11:28:53 -04:00
parent d314470d7f
commit 54b1fbed14
5 changed files with 330 additions and 3 deletions

View File

@@ -1,5 +1,39 @@
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Literal
FieldType = Literal["string", "password", "int", "bool", "textarea", "enum"]
@dataclass(frozen=True)
class ServiceConfigField:
"""
Declarative descriptor for one user-editable knob on a service.
The Inspector form (Fleet + MazeNET) renders inputs from this metadata,
and BaseService.validate_cfg coerces submitted values against it.
"""
key: str
label: str
type: FieldType = "string"
default: Any = None
secret: bool = False
help: str | None = None
enum: list[str] | None = None
placeholder: str | None = None
def to_json(self) -> dict:
d = asdict(self)
# Frontend doesn't need a None enum dangling on non-enum fields
if self.enum is None:
d.pop("enum", None)
return d
class ConfigValidationError(ValueError):
"""Raised when a submitted service_cfg value cannot be coerced to its declared type."""
class BaseService(ABC):
@@ -15,6 +49,10 @@ class BaseService(ABC):
default_image: str # Docker image tag, or "build" if a Dockerfile is needed
fleet_singleton: bool = False # True = runs once fleet-wide, not per-decky
# Per-service customizable fields exposed to the Inspector UI.
# Subclasses override; default empty -> "No customizable fields".
config_schema: list[ServiceConfigField] = []
@abstractmethod
def compose_fragment(
self,
@@ -41,3 +79,53 @@ class BaseService(ABC):
image built. Return None if default_image is used directly.
"""
return None
def validate_cfg(self, cfg: dict | None) -> dict:
"""
Coerce a user-submitted dict against this service's config_schema.
Unknown keys are silently dropped. Declared keys are coerced to their
declared type (raising ConfigValidationError on bad values). Empty
strings on optional fields drop the key entirely so compose_fragment's
existing `if "X" in cfg` guards keep working.
"""
out: dict[str, Any] = {}
if not cfg:
return out
by_key = {f.key: f for f in self.config_schema}
for key, raw in cfg.items():
spec = by_key.get(key)
if spec is None:
continue # drop unknown keys
if raw is None or raw == "":
continue
out[key] = _coerce(spec, raw)
return out
def _coerce(spec: ServiceConfigField, raw: Any) -> Any:
t = spec.type
if t in ("string", "password", "textarea"):
return str(raw)
if t == "int":
try:
return int(raw)
except (TypeError, ValueError) as e:
raise ConfigValidationError(f"{spec.key}: expected int, got {raw!r}") from e
if t == "bool":
if isinstance(raw, bool):
return raw
if isinstance(raw, str):
if raw.lower() in ("true", "1", "yes", "on"):
return True
if raw.lower() in ("false", "0", "no", "off"):
return False
raise ConfigValidationError(f"{spec.key}: expected bool, got {raw!r}")
if t == "enum":
s = str(raw)
if spec.enum and s not in spec.enum:
raise ConfigValidationError(
f"{spec.key}: {s!r} not in allowed values {spec.enum}"
)
return s
raise ConfigValidationError(f"{spec.key}: unknown field type {t!r}")

View File

@@ -1,6 +1,6 @@
import json
from pathlib import Path
from decnet.services.base import BaseService
from decnet.services.base import BaseService, ServiceConfigField
TEMPLATES_DIR = Path(__file__).parent.parent / "templates" / "http"
@@ -10,6 +10,41 @@ class HTTPService(BaseService):
ports = [80, 443]
default_image = "build"
config_schema = [
ServiceConfigField(
key="server_header",
label="Server header",
type="string",
placeholder="Apache/2.4.41 (Ubuntu)",
help="Value sent in the HTTP Server: response header.",
),
ServiceConfigField(
key="response_code",
label="Default response code",
type="int",
default=200,
),
ServiceConfigField(
key="fake_app",
label="Fake application",
type="enum",
enum=["none", "wordpress", "phpmyadmin", "tomcat", "jenkins"],
default="none",
help="Pre-baked application skin to render on the index page.",
),
ServiceConfigField(
key="extra_headers",
label="Extra headers (JSON or raw)",
type="textarea",
placeholder='{"X-Powered-By": "PHP/7.4.3"}',
),
ServiceConfigField(
key="custom_body",
label="Custom response body",
type="textarea",
),
]
def compose_fragment(
self,
decky_name: str,

View File

@@ -1,6 +1,6 @@
import json
from pathlib import Path
from decnet.services.base import BaseService
from decnet.services.base import BaseService, ServiceConfigField
TEMPLATES_DIR = Path(__file__).parent.parent / "templates" / "https"
@@ -10,6 +10,57 @@ class HTTPSService(BaseService):
ports = [443]
default_image = "build"
config_schema = [
ServiceConfigField(
key="server_header",
label="Server header",
type="string",
placeholder="nginx/1.18.0",
),
ServiceConfigField(
key="response_code",
label="Default response code",
type="int",
default=200,
),
ServiceConfigField(
key="fake_app",
label="Fake application",
type="enum",
enum=["none", "wordpress", "phpmyadmin", "tomcat", "jenkins"],
default="none",
),
ServiceConfigField(
key="extra_headers",
label="Extra headers (JSON or raw)",
type="textarea",
),
ServiceConfigField(
key="custom_body",
label="Custom response body",
type="textarea",
),
ServiceConfigField(
key="tls_cn",
label="TLS certificate CN",
type="string",
placeholder="mail.corp.local",
help="Common Name baked into the self-signed cert if no cert/key provided.",
),
ServiceConfigField(
key="tls_cert",
label="TLS certificate (PEM)",
type="textarea",
secret=True,
),
ServiceConfigField(
key="tls_key",
label="TLS private key (PEM)",
type="textarea",
secret=True,
),
]
def compose_fragment(
self,
decky_name: str,

View File

@@ -1,7 +1,7 @@
import os
from pathlib import Path
from decnet.services.base import BaseService
from decnet.services.base import BaseService, ServiceConfigField
TEMPLATES_DIR = Path(__file__).parent.parent / "templates" / "ssh"
ARTIFACTS_ROOT = os.environ.get("DECNET_ARTIFACTS_ROOT", "/var/lib/decnet/artifacts")
@@ -25,6 +25,27 @@ class SSHService(BaseService):
ports = [22]
default_image = "build"
config_schema = [
ServiceConfigField(
key="password",
label="Root password",
type="password",
default="admin",
secret=True,
help="Plaintext root password for the in-container sshd.",
),
ServiceConfigField(
key="hostname",
label="Container hostname",
type="string",
help=(
"Cosmetic override for the SSH banner/PS1 — keeps the decoy "
"looking heterogeneous. Decky identity (NODE_NAME) is unaffected."
),
placeholder="e.g. mail-01.corp.local",
),
]
def compose_fragment(
self,
decky_name: str,

View File

@@ -0,0 +1,132 @@
"""Schema-driven service config: descriptors, validation, compose round-trip."""
import pytest
from decnet.services.base import (
BaseService,
ConfigValidationError,
ServiceConfigField,
)
from decnet.services.http import HTTPService
from decnet.services.https import HTTPSService
from decnet.services.ssh import SSHService
class _Dummy(BaseService):
name = "dummy"
ports = [9999]
default_image = "alpine"
config_schema = [
ServiceConfigField(key="text", label="Text", type="string"),
ServiceConfigField(key="port", label="Port", type="int", default=8080),
ServiceConfigField(key="enabled", label="Enabled", type="bool"),
ServiceConfigField(
key="mode",
label="Mode",
type="enum",
enum=["a", "b", "c"],
),
ServiceConfigField(key="body", label="Body", type="textarea"),
ServiceConfigField(key="pw", label="Pw", type="password", secret=True),
]
def compose_fragment(self, decky_name, log_target=None, service_cfg=None):
return {"environment": dict(service_cfg or {})}
def test_unknown_keys_are_dropped():
cfg = _Dummy().validate_cfg({"text": "hi", "wat": "nope"})
assert cfg == {"text": "hi"}
def test_empty_string_drops_optional_key():
# compose_fragment guards on `if "key" in cfg`, so empty strings must
# not slip through as the literal "".
cfg = _Dummy().validate_cfg({"text": "", "port": 1234})
assert "text" not in cfg
assert cfg["port"] == 1234
def test_int_coercion_from_string():
cfg = _Dummy().validate_cfg({"port": "8443"})
assert cfg == {"port": 8443}
def test_int_rejects_garbage():
with pytest.raises(ConfigValidationError):
_Dummy().validate_cfg({"port": "eighty"})
def test_bool_coercion():
s = _Dummy()
assert s.validate_cfg({"enabled": "true"}) == {"enabled": True}
assert s.validate_cfg({"enabled": "0"}) == {"enabled": False}
assert s.validate_cfg({"enabled": True}) == {"enabled": True}
def test_enum_rejects_out_of_set():
with pytest.raises(ConfigValidationError):
_Dummy().validate_cfg({"mode": "z"})
def test_enum_accepts_valid():
assert _Dummy().validate_cfg({"mode": "b"}) == {"mode": "b"}
def test_none_cfg_returns_empty_dict():
assert _Dummy().validate_cfg(None) == {}
def test_field_to_json_omits_unused_enum():
f = ServiceConfigField(key="x", label="X", type="string")
assert "enum" not in f.to_json()
g = ServiceConfigField(key="m", label="M", type="enum", enum=["a", "b"])
assert g.to_json()["enum"] == ["a", "b"]
# --- Real services -----------------------------------------------------------
def test_ssh_schema_keys_match_compose_reads():
# SSHService.compose_fragment reads cfg.get("password") and cfg.get("hostname")
# — the schema must expose exactly those.
keys = {f.key for f in SSHService.config_schema}
assert keys == {"password", "hostname"}
def test_ssh_compose_round_trip_through_validator():
svc = SSHService()
cfg = svc.validate_cfg({"password": "hunter2", "hostname": "mail-01"})
frag = svc.compose_fragment("decky-test", service_cfg=cfg)
env = frag["environment"]
assert env["SSH_ROOT_PASSWORD"] == "hunter2"
assert env["SSH_HOSTNAME"] == "mail-01"
assert env["NODE_NAME"] == "decky-test"
def test_ssh_default_password_when_unset():
svc = SSHService()
cfg = svc.validate_cfg({})
frag = svc.compose_fragment("decky-test", service_cfg=cfg)
# Default fallback in compose_fragment is "admin"; validator returns {}
assert frag["environment"]["SSH_ROOT_PASSWORD"] == "admin"
def test_http_schema_covers_compose_keys():
keys = {f.key for f in HTTPService.config_schema}
# These are the keys HTTPService.compose_fragment branches on.
assert {"server_header", "response_code", "fake_app", "extra_headers", "custom_body"} <= keys
def test_http_response_code_int_coercion():
svc = HTTPService()
cfg = svc.validate_cfg({"response_code": "418"})
frag = svc.compose_fragment("decky-test", service_cfg=cfg)
assert frag["environment"]["RESPONSE_CODE"] == "418"
def test_https_schema_includes_tls_fields():
keys = {f.key for f in HTTPSService.config_schema}
assert {"tls_cn", "tls_cert", "tls_key"} <= keys
secrets = {f.key for f in HTTPSService.config_schema if f.secret}
assert {"tls_cert", "tls_key"} <= secrets