anti/DECNET
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add machine archetypes and amount= expansion

Browse Source 
Introduces archetype profiles (windows-workstation, linux-server,
domain-controller, printer, iot-device, etc.) so users get a realistic
service+distro combination without knowing which services to pick.

Adds amount= to INI config (and CLI --archetype) so a single section
can spawn N identical deckies without copy-paste. Per-service subsections
(e.g. [group.ssh]) propagate to all expanded instances automatically.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
anti
2026-04-04 13:08:01 -03:00
parent 9f93a8502c
commit bbb5fa7a7e
5 changed files with 569 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
decnet/archetypes.py Normal file
View File

@@ -0,0 +1,147 @@
"""
Machine archetype profiles for DECNET deckies.
An archetype is a pre-packaged identity: a realistic combination of services
and OS choices that makes a decky look like a specific class of machine
(workstation, printer, database server, etc.) without the user needing to
know which services or distros to pick.
Usage in INI config:
[my-workstations]
archetype=windows-workstation
amount=4
Usage via CLI:
decnet deploy --deckies 3 --archetype linux-server
"""
from __future__ import annotations
import random
from dataclasses import dataclass
@dataclass(frozen=True)
class Archetype:
slug: str
display_name: str
description: str
services: list[str] # default service set for this machine type
preferred_distros: list[str] # distro slugs to rotate through
ARCHETYPES: dict[str, Archetype] = {
"windows-workstation": Archetype(
slug="windows-workstation",
display_name="Windows Workstation",
description="Corporate Windows desktop: SMB shares + RDP access",
services=["smb", "rdp"],
preferred_distros=["debian", "ubuntu22"],
),
"windows-server": Archetype(
slug="windows-server",
display_name="Windows Server",
description="Windows domain member: SMB, RDP, and LDAP directory",
services=["smb", "rdp", "ldap"],
preferred_distros=["debian", "ubuntu22"],
),
"domain-controller": Archetype(
slug="domain-controller",
display_name="Domain Controller",
description="Active Directory DC: LDAP, SMB, RDP, LLMNR",
services=["ldap", "smb", "rdp", "llmnr"],
preferred_distros=["debian", "ubuntu22"],
),
"linux-server": Archetype(
slug="linux-server",
display_name="Linux Server",
description="General-purpose Linux host: SSH + HTTP",
services=["ssh", "http"],
preferred_distros=["debian", "ubuntu22", "rocky9", "fedora"],
),
"web-server": Archetype(
slug="web-server",
display_name="Web Server",
description="Public-facing web host: HTTP + FTP",
services=["http", "ftp"],
preferred_distros=["debian", "ubuntu22", "ubuntu20"],
),
"database-server": Archetype(
slug="database-server",
display_name="Database Server",
description="Data tier host: MySQL, PostgreSQL, Redis",
services=["mysql", "postgres", "redis"],
preferred_distros=["debian", "ubuntu22"],
),
"mail-server": Archetype(
slug="mail-server",
display_name="Mail Server",
description="SMTP/IMAP/POP3 mail relay",
services=["smtp", "pop3", "imap"],
preferred_distros=["debian", "ubuntu22"],
),
"file-server": Archetype(
slug="file-server",
display_name="File Server",
description="SMB/FTP/SFTP file storage node",
services=["smb", "ftp", "ssh"],
preferred_distros=["debian", "ubuntu22", "rocky9"],
),
"printer": Archetype(
slug="printer",
display_name="Network Printer",
description="Network-attached printer: SNMP + FTP",
services=["snmp", "ftp"],
preferred_distros=["alpine", "debian"],
),
"iot-device": Archetype(
slug="iot-device",
display_name="IoT Device",
description="Embedded/IoT device: MQTT, SNMP, Telnet",
services=["mqtt", "snmp", "telnet"],
preferred_distros=["alpine"],
),
"industrial-control": Archetype(
slug="industrial-control",
display_name="Industrial Control System",
description="ICS/SCADA node: Conpot (Modbus/S7/DNP3) + SNMP",
services=["conpot", "snmp"],
preferred_distros=["debian"],
),
"voip-server": Archetype(
slug="voip-server",
display_name="VoIP Server",
description="SIP PBX / VoIP gateway",
services=["sip"],
preferred_distros=["debian", "ubuntu22"],
),
"monitoring-node": Archetype(
slug="monitoring-node",
display_name="Monitoring Node",
description="Infrastructure monitoring host: SNMP + SSH",
services=["snmp", "ssh"],
preferred_distros=["debian", "rocky9"],
),
"devops-host": Archetype(
slug="devops-host",
display_name="DevOps Host",
description="CI/CD or container host: Docker API + SSH + K8s",
services=["docker_api", "ssh", "k8s"],
preferred_distros=["ubuntu22", "debian"],
),
}
def get_archetype(slug: str) -> Archetype:
if slug not in ARCHETYPES:
available = ", ".join(sorted(ARCHETYPES))
raise ValueError(f"Unknown archetype '{slug}'. Available: {available}")
return ARCHETYPES[slug]
def all_archetypes() -> dict[str, Archetype]:
return dict(ARCHETYPES)
def random_archetype() -> Archetype:
return random.choice(list(ARCHETYPES.values()))

72
decnet/cli.py
View File

@@ -15,6 +15,7 @@ import typer
from rich.console import Console
from rich.table import Table
from decnet.archetypes import Archetype, all_archetypes, get_archetype
from decnet.config import (
DeckyConfig,
DecnetConfig,
@@ -41,13 +42,16 @@ def _resolve_distros(
distros_explicit: list[str] | None,
randomize_distros: bool,
n: int,
archetype: Archetype | None = None,
) -> list[str]:
"""Return a list of n distro slugs based on CLI flags."""
"""Return a list of n distro slugs based on CLI flags or archetype preference."""
if distros_explicit:
# Round-robin the provided list to fill n slots
return [distros_explicit[i % len(distros_explicit)] for i in range(n)]
if randomize_distros:
return [random_distro().slug for _ in range(n)]
if archetype:
pool = archetype.preferred_distros
return [pool[i % len(pool)] for i in range(n)]
# Default: cycle through all distros to maximize heterogeneity
slugs = list(all_distros().keys())
return [slugs[i % len(slugs)] for i in range(n)]
@@ -60,10 +64,11 @@ def _build_deckies(
randomize_services: bool,
distros_explicit: list[str] | None = None,
randomize_distros: bool = False,
archetype: Archetype | None = None,
) -> list[DeckyConfig]:
deckies = []
used_combos: set[frozenset] = set()
distro_slugs = _resolve_distros(distros_explicit, randomize_distros, n)
distro_slugs = _resolve_distros(distros_explicit, randomize_distros, n, archetype)
for i, ip in enumerate(ips):
name = f"decky-{i + 1:02d}"
@@ -72,8 +77,9 @@ def _build_deckies(
if services_explicit:
svc_list = services_explicit
elif archetype:
svc_list = list(archetype.services)
elif randomize_services:
# Pick 1-3 random services from the full registry, avoid exact duplicates
svc_pool = _all_service_names()
attempts = 0
while True:
@@ -85,7 +91,7 @@ def _build_deckies(
svc_list = list(chosen)
used_combos.add(chosen)
else:
typer.echo("Error: provide --services or --randomize-services.", err=True)
typer.echo("Error: provide --services, --archetype, or --randomize-services.", err=True)
raise typer.Exit(1)
deckies.append(
@@ -97,6 +103,7 @@ def _build_deckies(
base_image=distro.image,
build_base=distro.build_base,
hostname=hostname,
archetype=archetype.slug if archetype else None,
)
)
return deckies
@@ -116,7 +123,6 @@ def _build_deckies_from_ini(
IPv4Address(s.ip) for s in ini.deckies if s.ip
}
# Build an IP iterator that skips reserved + explicit addresses
net = IPv4Network(subnet_cidr, strict=False)
reserved = {
net.network_address,
@@ -127,10 +133,20 @@ def _build_deckies_from_ini(
auto_pool = (str(addr) for addr in net.hosts() if addr not in reserved)
distro_slugs = _resolve_distros(None, randomize, len(ini.deckies))
deckies: list[DeckyConfig] = []
for i, spec in enumerate(ini.deckies):
distro = get_distro(distro_slugs[i])
for spec in ini.deckies:
# Resolve archetype (if any) — explicit services/distro override it
arch: Archetype | None = None
if spec.archetype:
try:
arch = get_archetype(spec.archetype)
except ValueError as e:
console.print(f"[red]{e}[/]")
raise typer.Exit(1)
# Distro: archetype preferred list → random → global cycle
distro_pool = arch.preferred_distros if arch else list(all_distros().keys())
distro = get_distro(distro_pool[len(deckies) % len(distro_pool)])
hostname = random_hostname(distro.slug)
ip = spec.ip or next(auto_pool, None)
@@ -149,6 +165,8 @@ def _build_deckies_from_ini(
)
raise typer.Exit(1)
svc_list = spec.services
elif arch:
svc_list = list(arch.services)
elif randomize:
svc_pool = _all_service_names()
count = random.randint(1, min(3, len(svc_pool)))
@@ -156,7 +174,7 @@ def _build_deckies_from_ini(
else:
console.print(
f"[red]Decky '[{spec.name}]' has no services= in config. "
"Add services= or use --randomize-services.[/]"
"Add services=, archetype=, or use --randomize-services.[/]"
)
raise typer.Exit(1)
@@ -168,6 +186,7 @@ def _build_deckies_from_ini(
base_image=distro.image,
build_base=distro.build_base,
hostname=hostname,
archetype=arch.slug if arch else None,
service_config=spec.service_config,
))
return deckies
@@ -186,6 +205,7 @@ def deploy(
randomize_distros: bool = typer.Option(False, "--randomize-distros", help="Assign a random distro to each decky"),
log_target: Optional[str] = typer.Option(None, "--log-target", help="Forward logs to ip:port (e.g. 192.168.1.5:5140)"),
log_file: Optional[str] = typer.Option(None, "--log-file", help="Write RFC 5424 syslog to this path inside containers (e.g. /var/log/decnet/decnet.log)"),
archetype_name: Optional[str] = typer.Option(None, "--archetype", "-a", help="Machine archetype slug (e.g. linux-server, windows-workstation)"),
dry_run: bool = typer.Option(False, "--dry-run", help="Generate compose file without starting containers"),
no_cache: bool = typer.Option(False, "--no-cache", help="Force rebuild all images, ignoring Docker layer cache"),
ipvlan: bool = typer.Option(False, "--ipvlan", help="Use IPvlan L2 instead of MACVLAN (required on WiFi interfaces)"),
@@ -255,8 +275,17 @@ def deploy(
console.print(f"[red]Unknown service(s): {unknown}. Available: {_all_service_names()}[/]")
raise typer.Exit(1)
if not services_list and not randomize_services:
console.print("[red]Specify --services or --randomize-services.[/]")
# Resolve archetype if provided
arch: Archetype | None = None
if archetype_name:
try:
arch = get_archetype(archetype_name)
except ValueError as e:
console.print(f"[red]{e}[/]")
raise typer.Exit(1)
if not services_list and not randomize_services and not arch:
console.print("[red]Specify --services, --archetype, or --randomize-services.[/]")
raise typer.Exit(1)
iface = interface or detect_interface()
@@ -283,6 +312,7 @@ def deploy(
decky_configs = _build_deckies(
deckies, ips, services_list, randomize_services,
distros_explicit=distros_list, randomize_distros=randomize_distros,
archetype=arch,
)
effective_log_target = log_target
effective_log_file = log_file
@@ -352,3 +382,21 @@ def list_distros() -> None:
for slug, profile in sorted(all_distros().items()):
table.add_row(slug, profile.display_name, profile.image)
console.print(table)
@app.command(name="archetypes")
def list_archetypes() -> None:
"""List all machine archetype profiles."""
table = Table(title="Machine Archetypes", show_lines=True)
table.add_column("Slug", style="bold cyan")
table.add_column("Display Name")
table.add_column("Default Services", style="green")
table.add_column("Description", style="dim")
for slug, arch in sorted(all_archetypes().items()):
table.add_row(
slug,
arch.display_name,
", ".join(arch.services),
arch.description,
)
console.print(table)

1
decnet/config.py
View File

@@ -26,6 +26,7 @@ class DeckyConfig(BaseModel):
base_image: str # Docker image for the base/IP-holder container
build_base: str = "debian:bookworm-slim" # apt-compatible image for service Dockerfiles
hostname: str
archetype: str | None = None # archetype slug if spawned from an archetype profile
service_config: dict[str, dict] = {} # optional per-service persona config
@field_validator("services")

54
decnet/ini_loader.py
View File

@@ -11,6 +11,8 @@ Format:
[hostname-1]
ip=192.168.1.82 # optional
services=ssh,smb # optional; falls back to --randomize-services
archetype=linux-server # optional; sets services+distros automatically
amount=3 # optional; spawn N deckies from this config (default: 1)
[hostname-1.ssh] # optional per-service persona config
kernel_version=5.15.0-76-generic
@@ -26,6 +28,11 @@ Format:
[hostname-3]
ip=192.168.1.32
# Archetype shorthand — spin up 5 windows workstations:
[corp-workstations]
archetype=windows-workstation
amount=5
# Custom (bring-your-own) service definitions:
[custom-myservice]
binary=my-docker-image:latest
@@ -44,6 +51,7 @@ class DeckySpec:
name: str
ip: str | None = None
services: list[str] | None = None
archetype: str | None = None
service_config: dict[str, dict] = field(default_factory=dict)
@@ -104,18 +112,54 @@ def load_ini(path: str | Path) -> IniConfig:
ip = s.get("ip")
svc_raw = s.get("services")
services = [sv.strip() for sv in svc_raw.split(",")] if svc_raw else None
cfg.deckies.append(DeckySpec(name=section, ip=ip, services=services))
archetype = s.get("archetype")
amount_raw = s.get("amount", "1")
try:
amount = int(amount_raw)
if amount < 1:
raise ValueError
except ValueError:
raise ValueError(f"[{section}] amount= must be a positive integer, got '{amount_raw}'")
if amount == 1:
cfg.deckies.append(DeckySpec(
name=section, ip=ip, services=services, archetype=archetype,
))
else:
# Expand into N deckies; explicit ip is ignored (can't share one IP)
if ip:
raise ValueError(
f"[{section}] Cannot combine ip= with amount={amount}. "
"Remove ip= or use amount=1."
)
for idx in range(1, amount + 1):
cfg.deckies.append(DeckySpec(
name=f"{section}-{idx:02d}",
ip=None,
services=services,
archetype=archetype,
))
# Second pass: collect per-service subsections [decky-name.service]
decky_names = {d.name for d in cfg.deckies}
# Also propagates to expanded deckies: [group.ssh] applies to group-01, group-02, ...
decky_map = {d.name: d for d in cfg.deckies}
for section in cp.sections():
if "." not in section:
continue
decky_name, _, svc_name = section.partition(".")
if decky_name not in decky_names:
continue # orphaned subsection — ignore
svc_cfg = {k: v for k, v in cp[section].items()}
decky_map[decky_name].service_config[svc_name] = svc_cfg
if decky_name in decky_map:
# Direct match — single decky
decky_map[decky_name].service_config[svc_name] = svc_cfg
else:
# Try to find expanded deckies with prefix "{decky_name}-NN"
matched = [
d for d in cfg.deckies
if d.name.startswith(f"{decky_name}-")
]
if not matched:
continue # orphaned subsection — ignore
for d in matched:
d.service_config[svc_name] = svc_cfg
return cfg

312
tests/test_archetypes.py Normal file
View File

@@ -0,0 +1,312 @@
"""
Tests for machine archetypes and the amount= expansion feature.
"""
from __future__ import annotations
import textwrap
import tempfile
import os
import pytest
from decnet.archetypes import (
ARCHETYPES,
all_archetypes,
get_archetype,
random_archetype,
)
from decnet.ini_loader import load_ini, DeckySpec
from decnet.distros import DISTROS
# ---------------------------------------------------------------------------
# Archetype registry
# ---------------------------------------------------------------------------
def test_all_archetypes_returns_all():
result = all_archetypes()
assert isinstance(result, dict)
assert len(result) == len(ARCHETYPES)
def test_get_archetype_known():
arch = get_archetype("linux-server")
assert arch.slug == "linux-server"
assert "ssh" in arch.services
def test_get_archetype_unknown_raises():
with pytest.raises(ValueError, match="Unknown archetype"):
get_archetype("does-not-exist")
def test_random_archetype_returns_valid():
arch = random_archetype()
assert arch.slug in ARCHETYPES
def test_every_archetype_has_services():
for slug, arch in ARCHETYPES.items():
assert arch.services, f"Archetype '{slug}' has no services"
def test_every_archetype_has_preferred_distros():
for slug, arch in ARCHETYPES.items():
assert arch.preferred_distros, f"Archetype '{slug}' has no preferred_distros"
def test_every_archetype_preferred_distro_is_valid():
valid_slugs = set(DISTROS.keys())
for slug, arch in ARCHETYPES.items():
for d in arch.preferred_distros:
assert d in valid_slugs, (
f"Archetype '{slug}' references unknown distro '{d}'"
)
# ---------------------------------------------------------------------------
# INI loader — archetype= parsing
# ---------------------------------------------------------------------------
def _write_ini(content: str) -> str:
"""Write INI content to a temp file and return the path."""
content = textwrap.dedent(content)
fd, path = tempfile.mkstemp(suffix=".ini")
os.write(fd, content.encode())
os.close(fd)
return path
def test_ini_archetype_parsed():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[my-server]
archetype=linux-server
""")
cfg = load_ini(path)
os.unlink(path)
assert len(cfg.deckies) == 1
assert cfg.deckies[0].archetype == "linux-server"
assert cfg.deckies[0].services is None # not overridden
def test_ini_archetype_with_explicit_services_override():
"""explicit services= must survive alongside archetype="""
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[my-server]
archetype=linux-server
services=ftp,smb
""")
cfg = load_ini(path)
os.unlink(path)
assert cfg.deckies[0].archetype == "linux-server"
assert cfg.deckies[0].services == ["ftp", "smb"]
# ---------------------------------------------------------------------------
# INI loader — amount= expansion
# ---------------------------------------------------------------------------
def test_ini_amount_one_keeps_section_name():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[my-printer]
archetype=printer
amount=1
""")
cfg = load_ini(path)
os.unlink(path)
assert len(cfg.deckies) == 1
assert cfg.deckies[0].name == "my-printer"
def test_ini_amount_expands_deckies():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[corp-ws]
archetype=windows-workstation
amount=5
""")
cfg = load_ini(path)
os.unlink(path)
assert len(cfg.deckies) == 5
for i, d in enumerate(cfg.deckies, start=1):
assert d.name == f"corp-ws-{i:02d}"
assert d.archetype == "windows-workstation"
assert d.ip is None # auto-allocated
def test_ini_amount_with_ip_raises():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[bad-group]
services=ssh
ip=10.0.0.50
amount=3
""")
with pytest.raises(ValueError, match="Cannot combine ip="):
load_ini(path)
os.unlink(path)
def test_ini_amount_invalid_value_raises():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[bad]
services=ssh
amount=potato
""")
with pytest.raises(ValueError, match="must be a positive integer"):
load_ini(path)
os.unlink(path)
def test_ini_amount_zero_raises():
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[bad]
services=ssh
amount=0
""")
with pytest.raises(ValueError, match="must be a positive integer"):
load_ini(path)
os.unlink(path)
def test_ini_amount_multiple_groups():
"""Two groups with different amounts expand independently."""
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[workers]
archetype=linux-server
amount=3
[printers]
archetype=printer
amount=2
""")
cfg = load_ini(path)
os.unlink(path)
assert len(cfg.deckies) == 5
names = [d.name for d in cfg.deckies]
assert names == ["workers-01", "workers-02", "workers-03", "printers-01", "printers-02"]
# ---------------------------------------------------------------------------
# INI loader — per-service subsections propagate to expanded deckies
# ---------------------------------------------------------------------------
def test_ini_subsection_propagates_to_expanded_deckies():
"""[group.ssh] must apply to group-01, group-02, ..."""
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[linux-hosts]
archetype=linux-server
amount=3
[linux-hosts.ssh]
kernel_version=5.15.0-76-generic
""")
cfg = load_ini(path)
os.unlink(path)
assert len(cfg.deckies) == 3
for d in cfg.deckies:
assert "ssh" in d.service_config
assert d.service_config["ssh"]["kernel_version"] == "5.15.0-76-generic"
def test_ini_subsection_direct_match_unaffected():
"""A direct [decky.svc] subsection must still work when amount=1."""
path = _write_ini("""
[general]
net=10.0.0.0/24
gw=10.0.0.1
[web-01]
services=http
[web-01.http]
server_header=Apache/2.4.51
""")
cfg = load_ini(path)
os.unlink(path)
assert cfg.deckies[0].service_config["http"]["server_header"] == "Apache/2.4.51"
# ---------------------------------------------------------------------------
# _build_deckies — archetype applied via CLI path
# ---------------------------------------------------------------------------
def test_build_deckies_archetype_sets_services():
from decnet.cli import _build_deckies
from decnet.archetypes import get_archetype
arch = get_archetype("mail-server")
result = _build_deckies(
n=2,
ips=["10.0.0.10", "10.0.0.11"],
services_explicit=None,
randomize_services=False,
archetype=arch,
)
assert len(result) == 2
for d in result:
assert set(d.services) == set(arch.services)
assert d.archetype == "mail-server"
def test_build_deckies_archetype_preferred_distros():
from decnet.cli import _build_deckies
from decnet.archetypes import get_archetype
arch = get_archetype("iot-device") # preferred_distros=["alpine"]
result = _build_deckies(
n=3,
ips=["10.0.0.10", "10.0.0.11", "10.0.0.12"],
services_explicit=None,
randomize_services=False,
archetype=arch,
)
for d in result:
assert d.distro == "alpine"
def test_build_deckies_explicit_services_override_archetype():
from decnet.cli import _build_deckies
from decnet.archetypes import get_archetype
arch = get_archetype("linux-server")
result = _build_deckies(
n=1,
ips=["10.0.0.10"],
services_explicit=["ftp"],
randomize_services=False,
archetype=arch,
)
assert result[0].services == ["ftp"]
assert result[0].archetype == "linux-server"