oss-forensics

1#!/usr/bin/env python32"""3OSS Forensics Evidence Store Manager4Manages a JSON-based evidence store for forensic investigations.5 6Commands:7  add      - Add a piece of evidence8  list     - List all evidence (optionally filter by type or actor)9  verify   - Re-check SHA-256 hashes for integrity10  query    - Search evidence by keyword11  export   - Export evidence as a Markdown table12  summary  - Print investigation statistics13 14Usage example:15  python3 evidence-store.py --store evidence.json add \16    --source "git fsck output" --content "dangling commit abc123" \17    --type git --actor "malicious-user" --url "https://github.com/owner/repo/commit/abc123"18 19  python3 evidence-store.py --store evidence.json list --type git20  python3 evidence-store.py --store evidence.json verify21  python3 evidence-store.py --store evidence.json export > evidence-table.md22"""23 24import json25import argparse26import os27import datetime28import hashlib29import sys30 31EVIDENCE_TYPES = [32    "git",           # Local git repository data (commits, reflog, fsck)33    "gh_api",        # GitHub REST API responses34    "gh_archive",    # GitHub Archive / BigQuery query results35    "web_archive",   # Wayback Machine snapshots36    "ioc",           # Indicator of Compromise (SHA, domain, IP, package name, etc.)37    "analysis",      # Derived analysis / cross-source correlation result38    "manual",        # Manually noted observation39    "vendor_report", # External security vendor report excerpt40]41 42VERIFICATION_STATES = ["unverified", "single_source", "multi_source_verified"]43 44IOC_TYPES = [45    "COMMIT_SHA", "FILE_PATH", "API_KEY", "SECRET", "IP_ADDRESS",46    "DOMAIN", "PACKAGE_NAME", "ACTOR_USERNAME", "MALICIOUS_URL",47    "WORKFLOW_FILE", "BRANCH_NAME", "TAG_NAME", "RELEASE_NAME", "OTHER",48]49 50 51def _now_iso():52    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds") + "Z"53 54 55def _sha256(content: str) -> str:56    return hashlib.sha256(content.encode("utf-8")).hexdigest()57 58 59class EvidenceStore:60    def __init__(self, filepath: str):61        self.filepath = filepath62        self.data = {63            "metadata": {64                "version": "2.0",65                "created_at": _now_iso(),66                "last_updated": _now_iso(),67                "investigation": "",68                "target_repo": "",69            },70            "evidence": [],71            "chain_of_custody": [],72        }73        if os.path.exists(filepath):74            try:75                with open(filepath, "r", encoding="utf-8") as f:76                    self.data = json.load(f)77            except (json.JSONDecodeError, IOError) as e:78                print(f"Error loading evidence store '{filepath}': {e}", file=sys.stderr)79                print("Hint: The file might be corrupted. Check for manual edits or syntax errors.", file=sys.stderr)80                sys.exit(1)81 82    def _save(self):83        self.data["metadata"]["last_updated"] = _now_iso()84        with open(self.filepath, "w", encoding="utf-8") as f:85            json.dump(self.data, f, indent=2, ensure_ascii=False)86 87    def _next_id(self) -> str:88        return f"EV-{len(self.data['evidence']) + 1:04d}"89 90    def add(91        self,92        source: str,93        content: str,94        evidence_type: str,95        actor: str = None,96        url: str = None,97        timestamp: str = None,98        ioc_type: str = None,99        verification: str = "unverified",100        notes: str = None,101    ) -> str:102        evidence_id = self._next_id()103        entry = {104            "id": evidence_id,105            "type": evidence_type,106            "source": source,107            "content": content,108            "content_sha256": _sha256(content),109            "actor": actor,110            "url": url,111            "event_timestamp": timestamp,112            "collected_at": _now_iso(),113            "ioc_type": ioc_type,114            "verification": verification,115            "notes": notes,116        }117        self.data["evidence"].append(entry)118        self.data["chain_of_custody"].append({119            "action": "add",120            "evidence_id": evidence_id,121            "timestamp": _now_iso(),122            "source": source,123        })124        self._save()125        return evidence_id126 127    def list_evidence(self, filter_type: str = None, filter_actor: str = None):128        results = self.data["evidence"]129        if filter_type:130            results = [e for e in results if e.get("type") == filter_type]131        if filter_actor:132            results = [e for e in results if e.get("actor") == filter_actor]133        return results134 135    def verify_integrity(self):136        """Re-compute SHA-256 for all entries and report mismatches."""137        issues = []138        for entry in self.data["evidence"]:139            expected = _sha256(entry["content"])140            stored = entry.get("content_sha256", "")141            if expected != stored:142                issues.append({143                    "id": entry["id"],144                    "stored_sha256": stored,145                    "computed_sha256": expected,146                })147        return issues148 149    def query(self, keyword: str):150        """Search for keyword in content, source, actor, or url."""151        keyword_lower = keyword.lower()152        return [153            e for e in self.data["evidence"]154            if keyword_lower in (e.get("content", "") or "").lower()155            or keyword_lower in (e.get("source", "") or "").lower()156            or keyword_lower in (e.get("actor", "") or "").lower()157            or keyword_lower in (e.get("url", "") or "").lower()158        ]159 160    def export_markdown(self) -> str:161        lines = [162            "# Evidence Registry",163            "",164            f"**Store**: `{self.filepath}`",165            f"**Last Updated**: {self.data['metadata'].get('last_updated', 'N/A')}",166            f"**Total Evidence Items**: {len(self.data['evidence'])}",167            "",168            "| ID | Type | Source | Actor | Verification | Event Timestamp | URL |",169            "|----|------|--------|-------|--------------|-----------------|-----|",170        ]171        for e in self.data["evidence"]:172            url = e.get("url") or ""173            url_display = f"[link]({url})" if url else ""174            lines.append(175                f"| {e['id']} | {e.get('type','')} | {e.get('source','')} "176                f"| {e.get('actor') or ''} | {e.get('verification','')} "177                f"| {e.get('event_timestamp') or ''} | {url_display} |"178            )179        lines.append("")180        lines.append("## Chain of Custody")181        lines.append("")182        lines.append("| Evidence ID | Action | Timestamp | Source |")183        lines.append("|-------------|--------|-----------|--------|")184        for c in self.data["chain_of_custody"]:185            lines.append(186                f"| {c.get('evidence_id','')} | {c.get('action','')} "187                f"| {c.get('timestamp','')} | {c.get('source','')} |"188            )189        return "\n".join(lines)190 191    def summary(self) -> dict:192        by_type = {}193        by_verification = {}194        actors = set()195        for e in self.data["evidence"]:196            t = e.get("type", "unknown")197            by_type[t] = by_type.get(t, 0) + 1198            v = e.get("verification", "unverified")199            by_verification[v] = by_verification.get(v, 0) + 1200            if e.get("actor"):201                actors.add(e["actor"])202        return {203            "total": len(self.data["evidence"]),204            "by_type": by_type,205            "by_verification": by_verification,206            "unique_actors": sorted(actors),207        }208 209 210def main():211    parser = argparse.ArgumentParser(212        description="OSS Forensics Evidence Store Manager v2.0",213        formatter_class=argparse.RawDescriptionHelpFormatter,214    )215    parser.add_argument("--store", default="evidence.json", help="Path to evidence JSON file (default: evidence.json)")216 217    subparsers = parser.add_subparsers(dest="command", metavar="COMMAND")218 219    # --- add ---220    add_p = subparsers.add_parser("add", help="Add a new evidence entry")221    add_p.add_argument("--source", required=True, help="Where this evidence came from (e.g. 'git fsck', 'GH API /commits')")222    add_p.add_argument("--content", required=True, help="The evidence content (commit SHA, API response excerpt, etc.)")223    add_p.add_argument("--type", required=True, choices=EVIDENCE_TYPES, dest="evidence_type", help="Evidence type")224    add_p.add_argument("--actor", help="GitHub handle or email of associated actor")225    add_p.add_argument("--url", help="URL to original source")226    add_p.add_argument("--timestamp", help="When the event occurred (ISO 8601)")227    add_p.add_argument("--ioc-type", choices=IOC_TYPES, help="IOC subtype (for --type ioc)")228    add_p.add_argument("--verification", choices=VERIFICATION_STATES, default="unverified")229    add_p.add_argument("--notes", help="Additional investigator notes")230    add_p.add_argument("--quiet", action="store_true", help="Suppress success message")231 232    # --- list ---233    list_p = subparsers.add_parser("list", help="List all evidence entries")234    list_p.add_argument("--type", dest="filter_type", choices=EVIDENCE_TYPES, help="Filter by type")235    list_p.add_argument("--actor", dest="filter_actor", help="Filter by actor")236 237    # --- verify ---238    subparsers.add_parser("verify", help="Verify SHA-256 integrity of all evidence content")239 240    # --- query ---241    query_p = subparsers.add_parser("query", help="Search evidence by keyword")242    query_p.add_argument("keyword", help="Keyword to search for")243 244    # --- export ---245    subparsers.add_parser("export", help="Export evidence as a Markdown table (stdout)")246 247    # --- summary ---248    subparsers.add_parser("summary", help="Print investigation statistics")249 250    args = parser.parse_args()251 252    if not args.command:253        parser.print_help()254        sys.exit(0)255 256    store = EvidenceStore(args.store)257 258    if args.command == "add":259        eid = store.add(260            source=args.source,261            content=args.content,262            evidence_type=args.evidence_type,263            actor=args.actor,264            url=args.url,265            timestamp=args.timestamp,266            ioc_type=args.ioc_type,267            verification=args.verification,268            notes=args.notes,269        )270        if not getattr(args, "quiet", False):271            print(f"✓ Added evidence: {eid}")272 273    elif args.command == "list":274        items = store.list_evidence(275            filter_type=getattr(args, "filter_type", None),276            filter_actor=getattr(args, "filter_actor", None),277        )278        if not items:279            print("No evidence found.")280        for e in items:281            actor_str = f" | actor: {e['actor']}" if e.get("actor") else ""282            url_str = f" | {e['url']}" if e.get("url") else ""283            print(f"[{e['id']}] {e['type']:12s} | {e['verification']:20s} | {e['source']}{actor_str}{url_str}")284 285    elif args.command == "verify":286        issues = store.verify_integrity()287        if not issues:288            print(f"✓ All {len(store.data['evidence'])} evidence entries passed SHA-256 integrity check.")289        else:290            print(f"✗ {len(issues)} integrity issue(s) detected:")291            for i in issues:292                print(f"  [{i['id']}] stored={i['stored_sha256'][:16]}... computed={i['computed_sha256'][:16]}...")293            sys.exit(1)294 295    elif args.command == "query":296        results = store.query(args.keyword)297        print(f"Found {len(results)} result(s) for '{args.keyword}':")298        for e in results:299            print(f"  [{e['id']}] {e['type']} | {e['source']} | {e['content'][:80]}")300 301    elif args.command == "export":302        print(store.export_markdown())303 304    elif args.command == "summary":305        s = store.summary()306        print(f"Total evidence items : {s['total']}")307        print(f"By type              : {json.dumps(s['by_type'], indent=2)}")308        print(f"By verification      : {json.dumps(s['by_verification'], indent=2)}")309        print(f"Unique actors        : {s['unique_actors']}")310 311 312if __name__ == "__main__":313    main()314