import warnings warnings.filterwarnings('ignore', category=ResourceWarning) #!/usr/bin/env python3 """ driftwatch_lite.py — Enterprise Edge & AI-Procurement Readiness Inspector (v2.1.0) Zero external dependencies (Python 3.8+ Standard Library only). Usage: python3 driftwatch_lite.py --domain example.com [--timeout 8] [--json] """ import argparse import datetime import json import socket import ssl import sys import urllib.error import urllib.request from typing import Dict, Any, Tuple VERSION = "2.1.0" USER_AGENT = f"TitanDriftWatch-CLI/{VERSION} (+https://titancapitalforensics.com/tools/driftwatch_lite.py)" WEIGHTS: Dict[str, int] = { "tls_valid_and_modern": 20, "hsts_preload_ready": 20, "csp_enforced": 20, "security_headers": 15, "llms_txt_manifest": 25, } def inspect_tls(domain: str, timeout: int) -> Dict[str, Any]: ctx = ssl.create_default_context() try: with socket.create_connection((domain, 443), timeout=timeout) as sock: with ctx.wrap_socket(sock, server_hostname=domain) as ssock: cert = ssock.getpeercert() cipher = ssock.cipher() tls_ver = ssock.version() not_after_str = cert.get("notAfter", "") days_left = -1 if not_after_str: exp_dt = datetime.datetime.strptime(not_after_str, "%b %d %H:%M:%S %Y %Z") days_left = (exp_dt - datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)).days return { "ok": True, "tls_version": tls_ver, "cipher_suite": cipher[0] if cipher else "UNKNOWN", "cert_days_remaining": days_left, "points": WEIGHTS["tls_valid_and_modern"] if (tls_ver in ("TLSv1.3", "TLSv1.2") and days_left > 14) else 10, } except Exception as exc: return {"ok": False, "error": str(exc), "points": 0} def fetch_url(url: str, timeout: int) -> Tuple[int, Dict[str, str], str]: req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(req, timeout=timeout) as resp: headers = {k.lower(): v for k, v in resp.info().items()} body = resp.read(4096).decode("utf-8", errors="ignore") return resp.getcode(), headers, body except urllib.error.HTTPError as e: headers = {k.lower(): v for k, v in e.headers.items()} if e.headers else {} return e.code, headers, "" except Exception as e: return 0, {"error": str(e)}, "" def run_audit(domain: str, timeout: int = 8) -> Dict[str, Any]: clean_dom = domain.strip().lower().replace("https://", "").replace("http://", "").split("/")[0] tls_info = inspect_tls(clean_dom, timeout) status_code, headers, _ = fetch_url(f"https://{clean_dom}/", timeout) # 1. HSTS Check hsts_raw = headers.get("strict-transport-security", "") hsts_pts = 0 if "max-age=" in hsts_raw.lower(): hsts_pts = 12 if "includesubdomains" in hsts_raw.lower(): hsts_pts += 4 if "preload" in hsts_raw.lower(): hsts_pts += 4 # 2. CSP Check csp_raw = headers.get("content-security-policy", "") csp_pts = 0 if csp_raw: csp_pts = 20 if ("default-src" in csp_raw.lower() and "frame-ancestors" in csp_raw.lower() and "unsafe-eval" not in csp_raw.lower()) else 10 # 3. Auxiliary Security Headers aux_pts = 0 for h in ("x-frame-options", "x-content-type-options", "referrer-policy"): if h in headers: aux_pts += 5 # 4. /llms.txt AI-Procurement Manifest Check llms_code, llms_hdrs, llms_body = fetch_url(f"https://{clean_dom}/llms.txt", timeout) llms_pts = 0 llms_status = f"MISSING (HTTP {llms_code})" if llms_code == 200 and len(llms_body.strip()) > 32 and "= 85 else ("B" if total_score >= 70 else ("C" if total_score >= 50 else "F")), "checks": { "tls": tls_info, "http_status": status_code, "hsts": {"present": bool(hsts_raw), "value": hsts_raw[:80], "points": hsts_pts, "max": 20}, "csp": {"present": bool(csp_raw), "unsafe_inline": "unsafe-inline" in csp_raw.lower(), "points": csp_pts, "max": 20}, "auxiliary_headers": {"points": aux_pts, "max": 15}, "llms_txt": {"status": llms_status, "points": llms_pts, "max": 25}, }, "remediation_portal": f"https://titancapitalforensics.com/xray/?d={clean_dom}", } def main() -> int: parser = argparse.ArgumentParser(description="Titan DriftWatch Lite — Enterprise Edge & AI-Procurement Inspector") parser.add_argument("target", nargs="?", default=None, help="Target corporate domain (positional)") parser.add_argument("--domain", "-d", default="titancapitalforensics.com", help="Target corporate domain (e.g. stripe.com)") parser.add_argument("--timeout", "-t", type=int, default=8, help="Socket/HTTP timeout in seconds (default: 8)") parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON output for CI/CD pipelines") args = parser.parse_args() res = run_audit(args.target or args.domain, args.timeout) if args.json: print(json.dumps(res, indent=2)) return 0 print("=" * 72) print(f" TITAN DRIFTWATCH LITE v{VERSION} // LIVE INFRASTRUCTURE AUDIT") print("=" * 72) print(f" Target Domain : {res['domain']}") print(f" Timestamp (UTC) : {res['timestamp_utc']}") print(f" Composite Score : {res['composite_score_100']}/100 (Grade: {res['grade']})") print("-" * 72) t = res["checks"]["tls"] print(f" [1] TLS Handshake : {t.get('points',0):2d}/20 pts | {t.get('tls_version','FAIL')} | Expiry: {t.get('cert_days_remaining','N/A')}d") print(f" [2] HSTS Policy : {res['checks']['hsts']['points']:2d}/20 pts | {'PRESENT' if res['checks']['hsts']['present'] else 'MISSING'}") print(f" [3] CSP Header : {res['checks']['csp']['points']:2d}/20 pts | {'PRESENT' if res['checks']['csp']['present'] else 'MISSING'}") print(f" [4] Aux Hardening : {res['checks']['auxiliary_headers']['points']:2d}/15 pts | XFO / XCTO / Referrer-Policy") print(f" [5] /llms.txt Index : {res['checks']['llms_txt']['points']:2d}/25 pts | {res['checks']['llms_txt']['status']}") print("-" * 72) print(f" Interactive Report : {res['remediation_portal']}") print("=" * 72) return 0 if __name__ == "__main__": sys.exit(main())