232 lines
8.2 KiB
Python
232 lines
8.2 KiB
Python
"""Inventory loading and validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from .models import UPDATE_CLASSES, Coupling, Item
|
|
|
|
VALID_CATEGORIES = {
|
|
"video", "audio", "rf", "lighting", "plugin", "os", "network", "other",
|
|
}
|
|
VALID_CRITICALITY = {"showstopper", "high", "normal", "low"}
|
|
VALID_STATUS = {"confirmed", "unconfirmed"}
|
|
|
|
|
|
class InventoryError(ValueError):
|
|
pass
|
|
|
|
|
|
def resolve_paths(path: Path) -> list[Path]:
|
|
"""Inventory can be split across files.
|
|
|
|
``inventory.yaml`` holds what the user actually confirmed; sibling
|
|
``inventory-*.yaml`` files hold proposed or optional additions. Keeping
|
|
them separate means a whole tranche of guesses can be deleted in one go
|
|
without touching the verified list.
|
|
"""
|
|
path = Path(path)
|
|
if path.is_dir():
|
|
found = sorted(path.glob("inventory*.yaml"))
|
|
if not found:
|
|
raise InventoryError(f"no inventory*.yaml in {path}")
|
|
return found
|
|
if not path.exists():
|
|
raise InventoryError(f"inventory not found: {path}")
|
|
extra = sorted(
|
|
p for p in path.parent.glob("inventory-*.yaml") if p != path
|
|
)
|
|
return [path] + extra
|
|
|
|
|
|
def load(path: Path) -> list[Item]:
|
|
"""Load every inventory file and return the merged item list."""
|
|
items: list[Item] = []
|
|
seen: set[str] = set()
|
|
for f in resolve_paths(path):
|
|
items.extend(_load_one(f, seen))
|
|
return items
|
|
|
|
|
|
def _load_one(path: Path, seen: set[str]) -> list[Item]:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
raw = yaml.safe_load(fh) or {}
|
|
|
|
entries = raw.get("items")
|
|
if not isinstance(entries, list):
|
|
raise InventoryError(f"{path.name} must contain a top-level 'items:' list")
|
|
|
|
defaults = raw.get("defaults", {}) or {}
|
|
items: list[Item] = []
|
|
|
|
for i, entry in enumerate(entries):
|
|
if not isinstance(entry, dict):
|
|
raise InventoryError(f"{path.name} items[{i}] is not a mapping")
|
|
merged = {**defaults, **entry}
|
|
|
|
item_id = str(merged.get("id") or "").strip()
|
|
if not item_id:
|
|
raise InventoryError(f"{path.name} items[{i}] missing required 'id'")
|
|
if item_id in seen:
|
|
raise InventoryError(
|
|
f"duplicate item id {item_id!r} in {path.name} -- it is "
|
|
f"already defined in another inventory file"
|
|
)
|
|
seen.add(item_id)
|
|
|
|
category = str(merged.get("category", "other")).lower()
|
|
if category not in VALID_CATEGORIES:
|
|
raise InventoryError(
|
|
f"{item_id}: category {category!r} not one of {sorted(VALID_CATEGORIES)}"
|
|
)
|
|
criticality = str(merged.get("criticality", "normal")).lower()
|
|
if criticality not in VALID_CRITICALITY:
|
|
raise InventoryError(
|
|
f"{item_id}: criticality {criticality!r} not one of {sorted(VALID_CRITICALITY)}"
|
|
)
|
|
|
|
source = merged.get("source") or {"kind": "manual"}
|
|
if not isinstance(source, dict):
|
|
raise InventoryError(f"{item_id}: 'source' must be a mapping")
|
|
|
|
status = str(merged.get("status", "confirmed")).lower()
|
|
if status not in VALID_STATUS:
|
|
raise InventoryError(
|
|
f"{item_id}: status {status!r} not one of {sorted(VALID_STATUS)}"
|
|
)
|
|
|
|
update_class = str(merged.get("update_class", "normal")).lower()
|
|
if update_class not in UPDATE_CLASSES:
|
|
raise InventoryError(
|
|
f"{item_id}: update_class {update_class!r} not one of "
|
|
f"{sorted(UPDATE_CLASSES)}"
|
|
)
|
|
|
|
coupled: list[Coupling] = []
|
|
for c in merged.get("coupled_with") or []:
|
|
if isinstance(c, str):
|
|
raise InventoryError(
|
|
f"{item_id}: coupled_with entries must be mappings with "
|
|
f"'id' and 'relation', got the bare string {c!r}"
|
|
)
|
|
if not isinstance(c, dict) or "id" not in c or "relation" not in c:
|
|
raise InventoryError(
|
|
f"{item_id}: each coupled_with entry needs 'id' and "
|
|
f"'relation'"
|
|
)
|
|
significant = str(c.get("significant", "major")).lower()
|
|
if significant not in ("major", "minor", "any"):
|
|
raise InventoryError(
|
|
f"{item_id}: coupled_with 'significant' must be "
|
|
f"major, minor or any (got {significant!r})"
|
|
)
|
|
coupled.append(Coupling(
|
|
id=str(c["id"]),
|
|
relation=str(c["relation"]),
|
|
note=c.get("note"),
|
|
significant=significant,
|
|
auto=bool(c.get("auto", False)),
|
|
))
|
|
|
|
installed = merged.get("installed")
|
|
items.append(
|
|
Item(
|
|
id=item_id,
|
|
name=str(merged.get("name") or item_id),
|
|
vendor=str(merged.get("vendor") or "unknown"),
|
|
category=category,
|
|
installed=str(installed) if installed is not None else None,
|
|
criticality=criticality,
|
|
source=source,
|
|
risk_queries=[str(q) for q in (merged.get("risk_queries") or [])],
|
|
github_repo=merged.get("github_repo"),
|
|
check_url=merged.get("check_url"),
|
|
notes=merged.get("notes"),
|
|
tags=[str(t) for t in (merged.get("tags") or [])],
|
|
freeze=bool(merged.get("freeze", False)),
|
|
status=status,
|
|
coupled_with=coupled,
|
|
update_group=merged.get("update_group"),
|
|
update_class=update_class,
|
|
expires=str(merged["expires"]) if merged.get("expires") else None,
|
|
version_hint=merged.get("version_hint"),
|
|
source_file=path.name,
|
|
)
|
|
)
|
|
|
|
return items
|
|
|
|
|
|
_ID_LINE = re.compile(r"^(\s*)-\s+id:\s*(\S+)\s*$")
|
|
_INSTALLED_LINE = re.compile(r"^(\s*)installed:.*$")
|
|
_NEW_ITEM = re.compile(r"^\s*-\s+id:\s*\S+")
|
|
|
|
|
|
def save_installed(path: Path, item_id: str, version: str) -> None:
|
|
"""Set one item's ``installed`` value by editing that single line.
|
|
|
|
Deliberately a text edit rather than a YAML load-and-dump: round-tripping
|
|
through PyYAML discards every comment in the file, and the comments in
|
|
inventory.yaml (which vendors cannot be scraped, which products still need
|
|
identifying, why a given item is marked showstopper) are a large part of
|
|
its value.
|
|
"""
|
|
# The item may live in any of the inventory files.
|
|
target: Path | None = None
|
|
lines: list[str] = []
|
|
start = None
|
|
indent = ""
|
|
|
|
for candidate in resolve_paths(Path(path)):
|
|
candidate_lines = candidate.read_text(encoding="utf-8").splitlines(
|
|
keepends=True)
|
|
for i, line in enumerate(candidate_lines):
|
|
m = _ID_LINE.match(line.rstrip("\n"))
|
|
if m and m.group(2).strip("'\"") == item_id:
|
|
target, lines, start = candidate, candidate_lines, i
|
|
indent = m.group(1) + " "
|
|
break
|
|
if target is not None:
|
|
break
|
|
|
|
if target is None or start is None:
|
|
raise InventoryError(f"no such item id: {item_id}")
|
|
path = target
|
|
|
|
# Quoted so values like 10.2 or 7.16 are not read back as floats.
|
|
replacement = f"installed: '{version}'"
|
|
|
|
# Walk to the end of this item's block.
|
|
end = len(lines)
|
|
for j in range(start + 1, len(lines)):
|
|
if _NEW_ITEM.match(lines[j]):
|
|
end = j
|
|
break
|
|
|
|
for j in range(start + 1, end):
|
|
im = _INSTALLED_LINE.match(lines[j].rstrip("\n"))
|
|
if im:
|
|
lines[j] = f"{im.group(1)}{replacement}\n"
|
|
break
|
|
else:
|
|
# No installed: key on this item -- insert one right after the id.
|
|
lines.insert(start + 1, f"{indent}{replacement}\n")
|
|
|
|
path.write_text("".join(lines), encoding="utf-8")
|
|
|
|
|
|
def by_id(items: list[Item]) -> dict[str, Item]:
|
|
return {i.id: i for i in items}
|
|
|
|
|
|
def load_file_ids(path: Path) -> list[str]:
|
|
"""Ids declared in one inventory file, for reporting which file an item
|
|
came from."""
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
raw = yaml.safe_load(fh) or {}
|
|
return [str(e.get("id")) for e in (raw.get("items") or [])
|
|
if isinstance(e, dict) and e.get("id")]
|