#!/usr/bin/env python3 """Stack inventory check for the marketing-mcp-setup skill. Usage (from the skill folder): python3 -I scripts/mcp_inventory_check.py INVENTORY.md [--as-of YYYY-MM-DD] [--json] Validates a filled assets/stack-inventory-template.md, one block per server, environment and client: C01 every block has all template labels, none of them blank, and a valid Status C02 no <...> placeholder remains C03 T1 to T5 carry PASS, WARN, FAIL or UNPROVEN, with "because" for WARN and UNPROVEN; T6 is present and, on a connected block, has text and a date C04 Status connected needs T1 PASS or WARN, T2 and T3 PASS, T4 PASS or UNPROVEN C05 a FAIL or blocked block is named under "Failed or blocked"; a T1 WARN under "Open risks" C06 a client-only read-only layer records "user told" with a date C07 Prepared has a date; Next review is at most 30 days after it, not overdue C08 T1 to T4 dates on connected blocks are at most 30 days old C09 "Connected and tested" equals the number of connected blocks C10 the heading names the client: ### Vendor (account, environment, client) --as-of defaults to today; pass it only to reproduce a dated example. Standard library only. No network calls, no subprocesses, no file writes. Exit codes: 0 no FAIL, 1 at least one FAIL, 2 bad input or unreadable file. """ import argparse import datetime as dt import json import os import re import sys ORDER = {"FAIL": 0, "WARN": 1} DATE_RE = re.compile(r"\b(\d{4}-\d{2}-\d{2})\b") REQUIRED_LABELS = [ "Status", "Job", "Route", "Endpoint or package", "Clients and config", "Identity", "Reachable accounts", "Granted scopes or permissions", "Read-only layer", "Write tools and how they are blocked", "Untrusted-content and send-out tools", "Limits and meters", "Tests", "Owner", "Revoke by", "Next review", "Agent access line for stack.md", ] STATUSES = ("connected", "blocked", "pending", "removed", "not connected") PLACEHOLDER = re.compile(r"<[^<>\n]{1,80}>") RESULT = re.compile(r"\b(PASS|WARN|FAIL|UNPROVEN)\b") COUNT_LABELS = ("connected and tested", "servers connected and tested") class InputError(Exception): """Bad arguments or unreadable input: exit code 2.""" def plural(n, word): return f"{n} {word}" if n == 1 else f"{n} {word}s" def parse_date(value, label): try: return dt.date.fromisoformat(value) except (TypeError, ValueError): raise InputError(f"{label} must be YYYY-MM-DD, got {value!r}") def parse_inventory(text): doc = {"prepared": None, "summary": {}, "servers": [], "placeholders": []} section, server, in_tests, in_fence = None, None, False, False for number, line in enumerate(text.splitlines(), 1): if line.strip().startswith("```"): in_fence = not in_fence continue if in_fence: continue if PLACEHOLDER.search(line): doc["placeholders"].append(number) if line.startswith("## "): section, server, in_tests = line[3:].strip().lower(), None, False continue if line.startswith("### ") and section and section.startswith("servers"): server = {"heading": line[4:].strip(), "line": number, "labels": {}, "tests": {}} doc["servers"].append(server) in_tests = False continue label = re.match(r"^- ([^:]+):\s*(.*)$", line) if section is None and label and label.group(1).strip() == "Prepared": found = DATE_RE.search(label.group(2)) doc["prepared"] = found.group(1) if found else None if section and section.startswith("summary") and label: doc["summary"][label.group(1).strip().lower()] = label.group(2).strip() if server is None: continue if label: server["labels"][label.group(1).strip()] = label.group(2).strip() in_tests = label.group(1).strip() == "Tests" continue test = re.match(r"^\s+-\s*(T[1-6])\b(.*)$", line) if test and in_tests: server["tests"][test.group(1)] = {"text": test.group(2).strip(), "line": number} return doc def check(args): try: with open(args.inventory, encoding="utf-8") as handle: text = handle.read() except OSError as exc: raise InputError(f"cannot read {args.inventory}: {exc}") except UnicodeDecodeError: raise InputError(f"cannot read {args.inventory}: it is not UTF-8 text, so it is not a filled inventory") as_of = parse_date(args.as_of, "--as-of") if args.as_of else dt.date.today() doc = parse_inventory(text) findings = [] def add(severity, rule, where, message): findings.append({"severity": severity, "rule": rule, "where": where, "message": message}) if not doc["servers"]: add("FAIL", "C01", "document", "no '### ' blocks under '## Servers'.") if doc["placeholders"]: shown = ", ".join(str(n) for n in doc["placeholders"][:12]) more = " and more" if len(doc["placeholders"]) > 12 else "" add("FAIL", "C02", "document", f"unfilled <...> placeholders on lines {shown}{more}.") prepared = parse_date(doc["prepared"], "Prepared") if doc["prepared"] else None if not prepared: add("FAIL", "C07", "document", "the header has no 'Prepared: YYYY-MM-DD' line.") start = prepared or as_of failed_line = doc["summary"].get("failed or blocked", "").lower() risks_line = doc["summary"].get("open risks", "").lower() connected = 0 for server in doc["servers"]: where = server["heading"] short = where.split(" (")[0].strip().lower() labels, tests = server["labels"], server["tests"] inner = re.search(r"\(([^()]*)\)\s*$", where) if not inner or len([p for p in inner.group(1).split(",") if p.strip()]) < 3: add("WARN", "C10", where, "the heading should name account, environment and client: " "### Vendor (account, environment, client).") missing = [label for label in REQUIRED_LABELS if label not in labels] if missing: add("FAIL", "C01", where, "missing labels: " + ", ".join(missing) + ".") blank = [label for label in REQUIRED_LABELS if label != "Tests" and label in labels and not labels[label]] if blank: add("FAIL", "C01", where, "empty labels: " + ", ".join(blank) + ".") status = labels.get("Status", "").lower() status_word = next((s for s in STATUSES if status.startswith(s)), None) if "Status" in labels and not status_word: add("FAIL", "C01", where, f"Status must start with one of: {', '.join(STATUSES)}.") results = {} for key in ("T1", "T2", "T3", "T4", "T5", "T6"): test = tests.get(key) if not test: add("FAIL", "C03", where, f"{key} is missing under Tests.") continue if key == "T6": detail = test["text"].partition(":")[2] if status_word == "connected" and not (detail.strip() and DATE_RE.search(detail)): add("FAIL", "C03", where, "T6 needs the revocation steps and the date of the disconnect and reconnect.") continue found = RESULT.search(test["text"]) if not found: add("FAIL", "C03", where, f"{key} has no PASS, WARN, FAIL or UNPROVEN result.") continue results[key] = found.group(1) if found.group(1) in ("UNPROVEN", "WARN") and "because" not in test["text"].lower(): add("FAIL", "C03", where, f"{key} is {found.group(1)} without 'because ...'.") if key in ("T1", "T2", "T3", "T4") and status_word == "connected": date = DATE_RE.search(test["text"]) if not date: add("FAIL", "C08", where, f"{key} has no test date.") else: age = (as_of - parse_date(date.group(1), key)).days if age > 30: add("FAIL", "C08", where, f"{key} is {age} days old; results expire after 30 days, so retest.") elif age < 0: add("FAIL", "C08", where, f"{key} is dated after {as_of.isoformat()}.") if status_word == "connected": connected += 1 rules = {"T1": ("PASS", "WARN"), "T2": ("PASS",), "T3": ("PASS",), "T4": ("PASS", "UNPROVEN")} for key, allowed in rules.items(): if key in results and results[key] not in allowed: add("FAIL", "C04", where, f"marked connected but {key} is {results[key]}; connected needs " f"{' or '.join(allowed)}.") if (any(r == "FAIL" for r in results.values()) or status_word == "blocked") and short not in failed_line: add("FAIL", "C05", where, "has a FAIL or is blocked but is not named under Summary 'Failed or blocked'.") if results.get("T1") == "WARN" and short not in risks_line: add("WARN", "C05", where, "T1 is WARN but the server is not named under Summary 'Open risks'.") layer = labels.get("Read-only layer", "").lower() if "client only" in layer and not ("user told" in layer and DATE_RE.search(layer)): add("FAIL", "C06", where, "a client-only read-only layer needs 'user told YYYY-MM-DD'.") review = DATE_RE.search(labels.get("Next review", "")) if "Next review" in labels: if not review: add("FAIL", "C07", where, "Next review has no date.") else: gap = (parse_date(review.group(1), "Next review") - start).days if gap > 30: add("FAIL", "C07", where, f"Next review is {gap} days after {start.isoformat()}; the limit is 30.") elif parse_date(review.group(1), "Next review") < as_of: add("FAIL", "C07", where, f"Next review {review.group(1)} is overdue.") stated_text = next((doc["summary"][k] for k in COUNT_LABELS if k in doc["summary"]), "") stated = re.match(r"^\s*(\d+)", stated_text) if doc["servers"] and (not stated or int(stated.group(1)) != connected): add("FAIL", "C09", "Summary", f"'Connected and tested' should be {connected}, the number of blocks with " f"Status connected.") findings.sort(key=lambda f: ORDER[f["severity"]]) counts = {k: sum(1 for f in findings if f["severity"] == k) for k in ORDER} if args.json: print(json.dumps({"as_of": as_of.isoformat(), "blocks": len(doc["servers"]), "connected": connected, "findings": findings, "summary": counts}, indent=2)) else: print(f"Inventory check of {os.path.basename(args.inventory)} as of {as_of.isoformat()}: " f"{plural(len(doc['servers']), 'block')}, {connected} connected. Nothing was changed.") for f in findings: print(f"{f['severity']} {f['rule']} {f['where']}: {f['message']}") if not findings: print("- no problems found") print(f"Summary: {counts['FAIL']} FAIL, {counts['WARN']} WARN") return 1 if counts["FAIL"] else 0 def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("inventory", help="the filled inventory Markdown file") parser.add_argument("--as-of", help="date for test age and review checks, YYYY-MM-DD (default: today)") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) try: return check(args) except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())