feat: add multi-worker concurrency and SARIF 2.1.0 export to audit_dir.py (#101) (#102)

service/scripts/audit_dir.py

@@ -10,11 +10,12 @@ from __future__ import annotations
 import argparse
 import os
 import sys
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from audit_lib import aggregate, print_human_report, scan_file
+from audit_lib import aggregate, format_sarif, print_human_report, scan_file
 from common import MAX_INPUT_BYTES, emit_json, eprint
 
 DEFAULT_SKIP_DIRS = {
@@ -45,10 +46,39 @@ def walk_files(root: Path, skip_dirs: set[str]):
                 yield path
 
 
+def _scan_worker(path: Path, check_stylometry: bool) -> tuple[dict | None, dict | None]:
+    try:
+        if path.stat().st_size > MAX_INPUT_BYTES:
+            return None, {"path": str(path), "reason": "too large"}
+        return scan_file(path, check_stylometry=check_stylometry), None
+    except Exception as e:  # keep the audit going on one bad file
+        return None, {"path": str(path), "reason": str(e)}
+
+
 def main() -> int:
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument("path", type=Path, help="Directory to audit recursively")
-    p.add_argument("--json", action="store_true", help="Emit a JSON report")
+    p.add_argument(
+        "--format",
+        choices=["human", "json", "sarif"],
+        default="human",
+        help="Output format (default: human)",
+    )
+    p.add_argument(
+        "--json", action="store_true", help="Emit a JSON report (alias for --format json)"
+    )
+    p.add_argument(
+        "--sarif",
+        action="store_true",
+        help="Emit an OASIS SARIF 2.1.0 report (alias for --format sarif)",
+    )
+    p.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        default=min(32, (os.cpu_count() or 1) + 4),
+        help="Number of concurrent worker threads (default: CPU cores + 4)",
+    )
     p.add_argument(
         "--check-stylometry",
         action="store_true",
@@ -72,16 +102,29 @@ def main() -> int:
         if part:
             skip_dirs.add(part)
 
-    files = []
-    skipped = []
-    for path in walk_files(root, skip_dirs):
-        try:
-            if path.stat().st_size > MAX_INPUT_BYTES:
-                skipped.append({"path": str(path), "reason": "too large"})
-                continue
-            files.append(scan_file(path, check_stylometry=args.check_stylometry))
-        except Exception as e:  # keep the audit going on one bad file
-            skipped.append({"path": str(path), "reason": str(e)})
+    paths = list(walk_files(root, skip_dirs))
+    files: list[dict] = []
+    skipped: list[dict] = []
+
+    if args.jobs <= 1:
+        for path in paths:
+            f, s = _scan_worker(path, args.check_stylometry)
+            if f is not None:
+                files.append(f)
+            if s is not None:
+                skipped.append(s)
+    else:
+        with ThreadPoolExecutor(max_workers=args.jobs) as pool:
+            futures = [pool.submit(_scan_worker, p, args.check_stylometry) for p in paths]
+            for fut in futures:
+                f, s = fut.result()
+                if f is not None:
+                    files.append(f)
+                if s is not None:
+                    skipped.append(s)
+
+    files.sort(key=lambda x: str(x.get("path", "")))
+    skipped.sort(key=lambda x: str(x.get("path", "")))
 
     summary = aggregate(files)
     report = {
@@ -92,8 +135,17 @@ def main() -> int:
         "files": files,
     }
 
+    out_format = args.format
     if args.json:
+        out_format = "json"
+    elif args.sarif:
+        out_format = "sarif"
+
+    if out_format == "json":
         emit_json(report)
+    elif out_format == "sarif":
+        sarif_doc = format_sarif(report)
+        emit_json(sarif_doc)
     else:
         print_human_report(
             files,

service/scripts/audit_lib.py

@@ -6,6 +6,7 @@ aggregate summary can be computed and rendered consistently.
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import Any
 
@@ -172,3 +173,123 @@ def print_human_report(
     for item in files:
         for msg, conf in zip(item.get("findings", []), item.get("confidence", []), strict=False):
             print(f"  [{conf}] {item['path']}: {msg}")
+
+
+def format_sarif(report: dict[str, Any]) -> dict[str, Any]:
+    """Convert an aggregate audit report into OASIS SARIF 2.1.0 format."""
+    rules = [
+        {
+            "id": "AI-WATERMARK-C2PA",
+            "name": "C2PAManifestDetected",
+            "shortDescription": {"text": "C2PA / Content Credentials provenance manifest detected"},
+            "fullDescription": {
+                "text": "A C2PA provenance manifest or JUMBF metadata box was detected in the asset."
+            },
+            "defaultConfiguration": {"level": "error"},
+            "properties": {"tags": ["provenance", "c2pa", "watermark"]},
+        },
+        {
+            "id": "AI-WATERMARK-METADATA",
+            "name": "AIMetadataMarkerDetected",
+            "shortDescription": {"text": "AI generation metadata or provenance markers detected"},
+            "fullDescription": {
+                "text": "AI metadata markers or container generator tags were detected in the file."
+            },
+            "defaultConfiguration": {"level": "warning"},
+            "properties": {"tags": ["provenance", "ai-generated"]},
+        },
+        {
+            "id": "AI-WATERMARK-UNICODE-LAYER-A",
+            "name": "InvisibleUnicodeWatermarkCarrier",
+            "shortDescription": {
+                "text": "Suspicious invisible Unicode or zero-width watermark carriers detected"
+            },
+            "fullDescription": {
+                "text": "Invisible Unicode formatting characters or homoglyph spaces used as watermark carriers were found in the text."
+            },
+            "defaultConfiguration": {"level": "warning"},
+            "properties": {"tags": ["watermark", "unicode", "layer-a"]},
+        },
+        {
+            "id": "AI-STYLES-HIGH-PROBABILITY",
+            "name": "HighProbabilityAITextCadence",
+            "shortDescription": {
+                "text": "High-probability statistical & stylometric AI text cadence detected"
+            },
+            "fullDescription": {
+                "text": "Stylometric analysis flagged the text as highly likely to be machine-generated."
+            },
+            "defaultConfiguration": {"level": "note"},
+            "properties": {"tags": ["stylometry", "ai-text"]},
+        },
+    ]
+
+    results = []
+    root_str = report.get("root", "")
+
+    for item in report.get("files", []):
+        file_path = item.get("path", "")
+        if root_str:
+            try:
+                rel_uri = os.path.relpath(file_path, root_str).replace("\\", "/")
+            except Exception:
+                rel_uri = file_path.replace("\\", "/")
+        else:
+            rel_uri = file_path.replace("\\", "/")
+
+        findings = item.get("findings", [])
+        confidences = item.get("confidence", [])
+
+        for msg, conf in zip(findings, confidences, strict=False):
+            rule_id = "AI-WATERMARK-METADATA"
+            level = "warning"
+
+            if "c2pa" in msg.lower() or "jumbf" in msg.lower() or item.get("has_c2pa"):
+                rule_id = "AI-WATERMARK-C2PA"
+                level = "error"
+            elif "layer-a" in msg.lower():
+                rule_id = "AI-WATERMARK-UNICODE-LAYER-A"
+                level = "warning" if conf in ("confirmed", "probable") else "note"
+            elif "stylometry" in msg.lower():
+                rule_id = "AI-STYLES-HIGH-PROBABILITY"
+                level = "note"
+            elif conf == "confirmed":
+                level = "error"
+            elif conf == "informational":
+                level = "note"
+
+            results.append(
+                {
+                    "ruleId": rule_id,
+                    "level": level,
+                    "message": {"text": msg},
+                    "locations": [
+                        {
+                            "physicalLocation": {
+                                "artifactLocation": {
+                                    "uri": rel_uri,
+                                    "uriBaseId": "%SRCROOT%",
+                                }
+                            }
+                        }
+                    ],
+                }
+            )
+
+    return {
+        "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
+        "version": "2.1.0",
+        "runs": [
+            {
+                "tool": {
+                    "driver": {
+                        "name": "watermarks-remover",
+                        "version": "0.1.0",
+                        "informationUri": "https://github.com/guillaumemeyer/watermarks-remover",
+                        "rules": rules,
+                    }
+                },
+                "results": results,
+            }
+        ],
+    }