ops-warden/src/warden/routing/models.py

87 lines
3.9 KiB
Python
Raw Normal View History

"""Data model for routing catalog entries.
A `RouteEntry` is a pointer: it names the owner and the authoritative doc for a
credential need. Only the SSH lane (`warden_executes: true`) may carry an authored
`steps` block and a `cert_command` pattern every other entry is identifiers and
pointers only (the no-double-source rule, enforced in `catalog.py`).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class RouteEntry:
id: str
title: str
need_keywords: List[str]
owner_repo: str
subsystem: str
warden_executes: bool
wiki_ref: str
canon_ref: str
reviewed: str
status: str # "active" | "draft"
# SSH lane only — None/empty for routed (non-executed) needs.
steps: List[str] = field(default_factory=list)
cert_command: Optional[str] = None
# Structured handoff (WP-0014) — optional, allowed on any lane. These are
# *templates and pointers* the `warden access` assist layer renders (and, for
# exec_capable lanes, proxies). They are NOT authored procedure prose and they
# never carry a secret value — only placeholders (`<...>`) and field names.
# Validation in catalog.py enforces the no-secret-material rule on every one.
auth_method: Optional[str] = None # how the caller authenticates to the owner
path_template: Optional[str] = None # owner-side path with `<...>` placeholders
fetch_command: Optional[str] = None # command skeleton run *as the caller*
exec_capable: bool = False # may `warden access --fetch/--exec` proxy it
policy_ref: Optional[str] = None # flex-auth check the fetch path runs first
# Proxy lane semantics (WP-0014 T4):
# "secret" — read a value (gated by flex-auth secret-read; caller must already
# be authenticated; value transits via inherit-stdout or child env).
# "login" — interactive auth bootstrap (OIDC/MFA). No secret-read gate (you have
# no identity yet), no caller-auth precheck (the point is to get one),
# run interactively as the caller; warden never captures the token.
lane: str = "secret"
@property
def is_active(self) -> bool:
return self.status == "active"
@property
def has_handoff(self) -> bool:
"""True when structured assist fields are present (advisory richness)."""
return any((self.auth_method, self.path_template, self.fetch_command))
@property
def resolvable(self) -> bool:
"""True when `warden access --fetch` can run this lane with no further input.
A resolvable lane is active, exec_capable, and its fetch command (with the path
inlined) carries no unresolved ``<...>`` placeholder. Template lanes like the
generic ``openbao-api-key`` or the ``<domain>``-parameterized login are *not*
resolvable until an owner ships concrete names. Lets an automated caller know
whether ``--fetch`` will work *before* attempting it (whynot-design request).
"""
if not (self.is_active and self.exec_capable and self.fetch_command):
return False
blob = f"{self.fetch_command} {self.path_template or ''}"
return "<" not in blob and ">" not in blob
def match_score(self, tokens: List[str]) -> int:
"""Keyword-overlap score against need_keywords, title, and id.
Pure ranking helper no I/O, no external calls.
"""
haystack = set(k.lower() for k in self.need_keywords)
haystack.update(self.id.lower().replace("-", " ").split())
haystack.update(self.title.lower().replace("-", " ").split())
score = 0
for tok in tokens:
t = tok.lower()
if t in haystack:
score += 2
elif any(t in h or h in t for h in haystack):
score += 1
return score