Generated scripts

conjurer_musician/revoke_shares.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+import os
+import re
+import time
+import json
+import argparse
+import sys
+from pathlib import Path
+
+# — CONFIGURATION — adjust as needed —
+APACHE_LOG     = '/var/log/apache2/share_access.log'  # <- point to your share-specific log!
+SHARE_DIR      = Path('/var/www/html/share')
+METADATA_FILE  = SHARE_DIR / '.downloads.json'
+STATE_FILE     = SHARE_DIR / '.logpos'
+# Regex to match lines like: "GET /share/abcdef1234... HTTP/1.1"
+TOKEN_REGEX    = re.compile(r'"\s*GET\s+/share/([0-9a-f]{32})\s+HTTP/')
+
+def debug(msg, args):
+    if args.debug:
+        print(f"[DEBUG] {msg}", file=sys.stderr)
+
+def load_metadata(args):
+    if METADATA_FILE.exists():
+        data = json.loads(METADATA_FILE.read_text())
+        debug(f"Loaded metadata: {data}", args)
+        return data
+    return {}
+
+def save_metadata(md, args):
+    METADATA_FILE.write_text(json.dumps(md))
+    debug(f"Saved metadata: {md}", args)
+
+def load_state(args):
+    if STATE_FILE.exists():
+        pos = int(STATE_FILE.read_text())
+        debug(f"Loaded last log position: {pos}", args)
+        return pos
+    return 0
+
+def save_state(pos, args):
+    STATE_FILE.write_text(str(pos))
+    debug(f"Saved new log position: {pos}", args)
+
+def scan_log(since_pos, args):
+    try:
+        with open(APACHE_LOG, 'r') as f:
+            f.seek(since_pos)
+            lines = f.readlines()
+            newpos = f.tell()
+        debug(f"Read {len(lines)} new lines", args)
+        return lines, newpos
+    except Exception as e:
+        print(f"Error reading log: {e}", file=sys.stderr)
+        return [], since_pos
+
+def record_downloads(lines, metadata, args):
+    now = time.time()
+    for line in lines:
+        m = TOKEN_REGEX.search(line)
+        if m:
+            tok = m.group(1)
+            if tok not in metadata:
+                metadata[tok] = now
+                debug(f"Recorded download of token {tok} at {now}", args)
+    return metadata
+
+def revoke_old(metadata, args):
+    now = time.time()
+    to_remove = [tok for tok, ts in metadata.items() if now - ts >= 120]
+    for tok in to_remove:
+        link = SHARE_DIR / tok
+        try:
+            link.unlink()
+            debug(f"Unlinked token {tok}", args)
+        except FileNotFoundError:
+            debug(f"Link {link} not found when revoking", args)
+        metadata.pop(tok, None)
+    return bool(to_remove)
+
+def main():
+    p = argparse.ArgumentParser(description="Revoke share links 2min after download")
+    p.add_argument('--debug', action='store_true', help='print debug info to stderr')
+    args = p.parse_args()
+
+    SHARE_DIR.mkdir(parents=True, exist_ok=True)
+    metadata = load_metadata(args)
+    state    = load_state(args)
+
+    lines, newpos = scan_log(state, args)
+    metadata      = record_downloads(lines, metadata, args)
+    save_metadata(metadata, args)               # always persist new downloads
+
+    if revoke_old(metadata, args):
+        save_metadata(metadata, args)           # persist removals
+
+    save_state(newpos, args)
+
+if __name__ == '__main__':
+    main()