From 800f539ef6c72c60fcd5615eb00ad21507f45cec Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Mon, 6 Jul 2026 16:09:12 -0700 Subject: [PATCH] feat(core+api): sockets, consumption audit, capabilities, mesh endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Graph model gains, all derived (GET /graph auto-exposes via asdict): - endpoints[{protocol,port}] + reach per node — reusing http_exposed/tcp_port, so raw-TCP infra (postgres/neo4j/mqtt) is finally visible and edges can be protocol-typed by the target's socket. - base_url for reference (external) nodes. - provides/consumes capability types, activating the dormant ProgramSpec fields. New surfaces: - core/audit.py + GET /graph/suggestions: SUGGESTS undeclared `requires` by matching a deployment's env endpoint values against provider sockets. Never writes; the graph stays declaration-derived (docs/relationships.md). - kind-scoped PUT/DELETE /config/references/{name} for external resources. - GET /mesh/deployments (flattened remote deployments + derived endpoints), and the mesh MQTT payload now carries tcp_port + base_url (forward-compatible, still no secrets) so peers can resolve cross-node endpoints. --- castle-api/src/castle_api/config_editor.py | 12 ++++ castle-api/src/castle_api/graph.py | 9 +++ castle-api/src/castle_api/mqtt_client.py | 8 +++ castle-api/src/castle_api/nodes.py | 38 +++++++++++ core/src/castle_core/audit.py | 74 ++++++++++++++++++++++ core/src/castle_core/relations.py | 47 +++++++++++++- 6 files changed, 185 insertions(+), 3 deletions(-) create mode 100644 core/src/castle_core/audit.py diff --git a/castle-api/src/castle_api/config_editor.py b/castle-api/src/castle_api/config_editor.py index d92b3e9..b408b19 100644 --- a/castle-api/src/castle_api/config_editor.py +++ b/castle-api/src/castle_api/config_editor.py @@ -458,6 +458,18 @@ def delete_static(name: str) -> dict: return _delete_deployment(name, kind="static") +@router.put("/references/{name}") +def save_reference(name: str, request: ServiceConfigRequest) -> dict: + """Create/update an external *reference* (manager: none, base_url) — an + endpoint castle doesn't run (a SaaS API, a remote/external service).""" + return _save_deployment(name, request.config, kind="reference") + + +@router.delete("/references/{name}") +def delete_reference(name: str) -> dict: + return _delete_deployment(name, kind="reference") + + @router.post("/apply", response_model=ApplyResponse) async def apply_config() -> ApplyResponse: """Converge the running system to match castle.yaml (a thin wrapper on core diff --git a/castle-api/src/castle_api/graph.py b/castle-api/src/castle_api/graph.py index d74afd3..4885277 100644 --- a/castle-api/src/castle_api/graph.py +++ b/castle-api/src/castle_api/graph.py @@ -11,6 +11,7 @@ import dataclasses from fastapi import APIRouter +from castle_core.audit import suggest_consumption from castle_core.relations import build_model from castle_api.config import get_config @@ -28,3 +29,11 @@ def get_graph() -> dict: "nodes": [dataclasses.asdict(n) for n in model.nodes], "edges": [dataclasses.asdict(e) for e in model.edges], } + + +@graph_router.get("/graph/suggestions") +def get_suggestions() -> dict: + """Undeclared-consumption *suggestions* — an opt-in advisory that matches each + deployment's env endpoint values against provider sockets. Never writes; the + graph itself stays declaration-derived. Accept one by declaring the `requires`.""" + return {"suggestions": [dataclasses.asdict(s) for s in suggest_consumption(get_config())]} diff --git a/castle-api/src/castle_api/mqtt_client.py b/castle-api/src/castle_api/mqtt_client.py index 94d3492..d65651d 100644 --- a/castle-api/src/castle_api/mqtt_client.py +++ b/castle-api/src/castle_api/mqtt_client.py @@ -61,6 +61,12 @@ def _registry_to_json(registry: NodeRegistry) -> str: entry["schedule"] = comp.schedule if comp.managed: entry["managed"] = comp.managed + # Socket surface + external target — so a peer can resolve cross-node + # consumption endpoints (still no secrets: only ports/URLs). + if getattr(comp, "tcp_port", None) is not None: + entry["tcp_port"] = comp.tcp_port + if getattr(comp, "base_url", None): + entry["base_url"] = comp.base_url data["deployed"][NodeRegistry.key(comp.kind, name)] = entry return json.dumps(data) @@ -93,6 +99,8 @@ def _json_to_registry(payload: str) -> NodeRegistry: subdomain=comp_data.get("subdomain"), schedule=comp_data.get("schedule"), managed=comp_data.get("managed", False), + tcp_port=comp_data.get("tcp_port"), + base_url=comp_data.get("base_url"), ) return NodeRegistry(node=node, deployed=deployed) diff --git a/castle-api/src/castle_api/nodes.py b/castle-api/src/castle_api/nodes.py index c71751a..4d70b24 100644 --- a/castle-api/src/castle_api/nodes.py +++ b/castle-api/src/castle_api/nodes.py @@ -93,6 +93,44 @@ def list_nodes() -> list[NodeSummary]: return nodes +_TCP_PROTOCOL = {5432: "pg", 7687: "bolt", 1883: "mqtt", 6379: "redis"} + + +def _endpoints_of_registry(d: object) -> list[dict]: + """Derive display endpoints from a registry deployment (mirrors relations).""" + eps: list[dict] = [] + port = getattr(d, "port", None) + if port is not None: + eps.append({"protocol": "http", "port": port}) + tcp = getattr(d, "tcp_port", None) + if tcp is not None: + eps.append({"protocol": _TCP_PROTOCOL.get(tcp, "tcp"), "port": tcp}) + return eps + + +@router.get("/mesh/deployments") +def mesh_deployments() -> dict: + """Flattened remote (mesh-discovered) deployments with derived endpoints — the + data the System Map needs to render other machines. Local node excluded (it's + already in /graph). Cross-node dependency *edges* need `requires` in the mesh + payload, which the registry doesn't carry yet.""" + out: list[dict] = [] + for hostname, remote in mesh_state.all_nodes(include_stale=True).items(): + for _kind, name, d in remote.registry.all(): + out.append( + { + "name": name, + "kind": d.kind, + "node": hostname, + "port": d.port, + "base_url": getattr(d, "base_url", None), + "subdomain": d.subdomain, + "endpoints": _endpoints_of_registry(d), + } + ) + return {"deployments": out} + + @router.get("/nodes/{hostname}", response_model=NodeDetail) def get_node(hostname: str) -> NodeDetail: """Get detailed info for a specific node.""" diff --git a/core/src/castle_core/audit.py b/core/src/castle_core/audit.py new file mode 100644 index 0000000..bfd0c89 --- /dev/null +++ b/core/src/castle_core/audit.py @@ -0,0 +1,74 @@ +"""Consumption audit — *suggests* undeclared ``requires`` by matching a +deployment's env endpoint values against known provider sockets. + +This is the one place castle looks at env → dependency, and it does so **only to +propose a declaration the user confirms** — never to write, and never to feed the +graph or ``functional?``. The relationship graph stays strictly declaration-derived +(see docs/relationships.md, "Env is derived *from* requires, never scraped *into* +it"); this module is an explicit, opt-in lint that sits *on top* of it. A suggestion +is accepted by writing a real ``requires`` (a declaration), at which point it stops +being a suggestion. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from castle_core.config import CastleConfig +from castle_core.relations import build_model + +# Hosts that mean "a provider on this node" — a port match against them is a strong +# signal (ports are unique per host). 172.17.0.1 is the docker bridge to the host. +_LOCAL_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "::1", "172.17.0.1"} + +# scheme://host:port OR host:port — anywhere inside an env value. +_HOSTPORT = re.compile(r"(?:(?P[a-z][\w+.-]*)://)?(?P[\w.-]+):(?P\d{2,5})") + + +@dataclass +class Suggestion: + consumer: str # the deployment whose env references the endpoint + provider: str # the provider deployment the port resolves to + env_var: str # the env var that revealed it + endpoint: str # host:port as seen in the value + protocol: str # the provider's socket protocol + + +def suggest_consumption(config: CastleConfig) -> list[Suggestion]: + """Undeclared consumption suggestions, derived from env endpoint values. + + For each deployment, scan its ``defaults.env`` values for ``host:port`` (or a + URL). When the port resolves to a *local* provider's socket and the consumer + doesn't already declare it, propose the edge. Deduped per (consumer, provider).""" + model = build_model(config, check=False) + # port -> (provider name, protocol); ports are unique per host, so a port match + # against a local host is a confident resolution. + port_provider: dict[int, tuple[str, str]] = {} + for n in model.nodes: + for ep in n.endpoints: + port_provider.setdefault(ep.port, (n.name, ep.protocol)) + + out: list[Suggestion] = [] + for _kind, name, dep in config.all_deployments(): + env = dict(dep.defaults.env) if (dep.defaults and dep.defaults.env) else {} + declared = {r.ref for r in getattr(dep, "requires", [])} + proposed: set[str] = set() + for var, val in env.items(): + for m in _HOSTPORT.finditer(str(val)): + host = m.group("host") + port = int(m.group("port")) + prov = port_provider.get(port) + if not prov: + continue + pname, proto = prov + # Only resolve when the value points at a local host (this node's + # providers) or names the provider directly — avoids matching a + # coincidental external host that happens to share a port number. + if host not in _LOCAL_HOSTS and host != pname: + continue + if pname == name or pname in declared or pname in proposed: + continue + proposed.add(pname) + out.append(Suggestion(name, pname, var, f"{host}:{port}", proto)) + return out diff --git a/core/src/castle_core/relations.py b/core/src/castle_core/relations.py index 5a822c5..3dfb929 100644 --- a/core/src/castle_core/relations.py +++ b/core/src/castle_core/relations.py @@ -24,7 +24,7 @@ from pathlib import Path from castle_core import git from castle_core.config import CastleConfig -from castle_core.manifest import Requirement +from castle_core.manifest import Requirement, SystemdDeployment @dataclass @@ -53,6 +53,16 @@ class Edge: bind: str | None = None # env var to project the target URL into (deployment) +@dataclass +class Endpoint: + """A socket a deployment exposes. ``protocol`` is a display heuristic — the + manifest has no protocol field (expose is http XOR tcp), so raw TCP is refined + by well-known port (else ``tcp``). A real protocol field is future work.""" + + protocol: str # "http" | "tcp" | "pg" | "bolt" | "mqtt" | "redis" + port: int + + @dataclass class Node: name: str # deployment name @@ -64,6 +74,11 @@ class Node: functional: bool = True # derived: all requirements satisfied fresh: bool | None = None # derived: its repo is at latest + clean deployed: bool | None = None # derived: active in the registry (None = unknown) + reach: str | None = None # off|internal|public (systemd/caddy), else None + endpoints: list[Endpoint] = field(default_factory=list) # sockets it exposes + base_url: str | None = None # the target URL, for kind=="reference" (external) + provides: list[str] = field(default_factory=list) # capability types (from program) + consumes: list[str] = field(default_factory=list) # capability types (from program) @dataclass @@ -163,6 +178,25 @@ def _check(config: CastleConfig, req: Requirement) -> bool: return True +# Well-known TCP ports → a friendlier protocol label (display heuristic only). +_TCP_PROTOCOL = {5432: "pg", 7687: "bolt", 1883: "mqtt", 6379: "redis"} + + +def _endpoints_of(dep: object) -> list[Endpoint]: + """The sockets a deployment exposes, derived from its manifest — reusing the + ``http_exposed`` / ``tcp_port`` accessors (SystemdDeployment only). Tools, + statics, and references have no expose block and yield ``[]``.""" + eps: list[Endpoint] = [] + if not isinstance(dep, SystemdDeployment): + return eps # only systemd services/jobs carry an expose block + exp = dep.expose + if dep.http_exposed and exp and exp.http: + eps.append(Endpoint("http", exp.http.internal.port)) + if dep.tcp_port is not None: + eps.append(Endpoint(_TCP_PROTOCOL.get(dep.tcp_port, "tcp"), dep.tcp_port)) + return eps + + def build_model( config: CastleConfig, check: bool = True, @@ -206,11 +240,13 @@ def build_model( if check else [] ) - repo_key = repo_of.get(_program_of(name, dep)) + prog_name = _program_of(name, dep) + repo_key = repo_of.get(prog_name) + prog = config.programs.get(prog_name) if prog_name else None nodes.append( Node( name=name, - program=_program_of(name, dep), + program=prog_name, kind=_nk, repo=repo_key, depended_on_by=fan_in.get(name, 0), @@ -218,6 +254,11 @@ def build_model( functional=not unmet, fresh=fresh_of.get(repo_key) if (freshness and repo_key) else None, deployed=(name in active) if active is not None else None, + reach=getattr(getattr(dep, "reach", None), "value", None), + endpoints=_endpoints_of(dep), + base_url=getattr(dep, "base_url", None), + provides=[c.type for c in prog.provides] if prog else [], + consumes=[c.type for c in prog.consumes] if prog else [], ) ) return Model(repos=list(repos.values()), nodes=nodes, edges=edges)