#!/usr/bin/env python3 """Lumen Local Agent — opt-in local file access (F1499). WHAT THIS IS Lumen (ask.eliteaiempire.com) is a hosted answer engine. By default it CANNOT see any file on your computer — its sandbox is fully isolated. If you WANT Lumen to read or work with files in ONE folder you choose, run this script. It does three things and nothing else: 1. Exposes exactly ONE directory you pick — never anything outside it. 2. Polls Lumen for file jobs your own agent run has queued, runs each inside that directory, and returns the result. 3. Stays read-only unless you explicitly pass --allow-write. Lumen never connects INTO your machine. THIS script makes outbound HTTPS requests to ask.eliteaiempire.com only. Stop it any time (Ctrl-C) and local access ends immediately. USAGE python3 lumen-local-agent.py # share the current folder, read-only python3 lumen-local-agent.py --dir ~/work # share a specific folder python3 lumen-local-agent.py --dir ~/work --allow-write # allow writes too It prints a 6-digit PAIRING CODE. Sign in at ask.eliteaiempire.com, open Settings, and enter the code to bind this agent to your account for this session. Revoke any time from Settings (or just stop the script). NO DEPENDENCIES — standard library only. Works on macOS, Linux, Windows. CONSENT & DISCLAIMER Running this script is your explicit, informed consent to let Lumen access the folder you choose. It runs on YOUR machine under YOUR account. It is provided "AS IS", WITHOUT WARRANTY OF ANY KIND. To the maximum extent permitted by law, Elite AI Empire is not liable for any data loss or file modification resulting from its use. Only point it at folders and data you are authorised to access. Full terms: https://ask.eliteaiempire.com/local-agent and /legal/terms """ import argparse import json import os import random import sys import time import urllib.request LUMEN = os.environ.get("LUMEN_BASE", "https://ask.eliteaiempire.com") POLL_INTERVAL = 3.0 MAX_READ_BYTES = 256 * 1024 # never return more than 256 KB per read MAX_SEARCH_HITS = 60 UA = "LumenLocalAgent/1.0 (+https://ask.eliteaiempire.com)" def _post(path, payload): data = json.dumps(payload).encode() req = urllib.request.Request( LUMEN + path, data=data, method="POST", headers={"Content-Type": "application/json", "User-Agent": UA}) with urllib.request.urlopen(req, timeout=20) as r: return json.loads(r.read().decode("utf-8", "replace") or "{}") def _safe(base, rel): """Resolve `rel` under `base`; raise if it escapes the jail.""" rel = (rel or ".").strip() full = os.path.realpath(os.path.join(base, rel)) if full != base and not full.startswith(base + os.sep): raise PermissionError(f"path '{rel}' is outside the shared folder") return full def do_job(job, base, allow_write): op = job.get("op") if op == "list": full = _safe(base, job.get("path", ".")) if not os.path.isdir(full): return {"ok": False, "error": "not a directory"} entries = [] for name in sorted(os.listdir(full))[:500]: p = os.path.join(full, name) entries.append({"name": name, "type": "dir" if os.path.isdir(p) else "file", "size": (os.path.getsize(p) if os.path.isfile(p) else 0)}) return {"ok": True, "path": os.path.relpath(full, base), "entries": entries} if op == "stat": full = _safe(base, job.get("path", ".")) if not os.path.exists(full): return {"ok": False, "error": "not found"} st = os.stat(full) return {"ok": True, "path": os.path.relpath(full, base), "size": st.st_size, "mtime": st.st_mtime, "is_dir": os.path.isdir(full)} if op == "read": full = _safe(base, job.get("path", "")) if not os.path.isfile(full): return {"ok": False, "error": "not a file"} limit = min(int(job.get("max_bytes") or MAX_READ_BYTES), MAX_READ_BYTES) with open(full, "rb") as f: raw = f.read(limit + 1) truncated = len(raw) > limit text = raw[:limit].decode("utf-8", "replace") return {"ok": True, "path": os.path.relpath(full, base), "text": text, "truncated": truncated} if op == "search": query = (job.get("query") or "").strip() if not query: return {"ok": False, "error": "empty query"} glob = job.get("glob") or "" hits = [] for root, _dirs, files in os.walk(base): for fn in files: if glob and not fn.endswith(glob.lstrip("*")): continue fp = os.path.join(root, fn) try: if os.path.getsize(fp) > 2 * 1024 * 1024: continue with open(fp, "r", errors="ignore") as f: for i, line in enumerate(f, 1): if query.lower() in line.lower(): hits.append({ "path": os.path.relpath(fp, base), "line": i, "text": line.strip()[:200]}) if len(hits) >= MAX_SEARCH_HITS: return {"ok": True, "hits": hits, "truncated": True} except Exception: continue return {"ok": True, "hits": hits, "truncated": False} if op == "write": if not allow_write: return {"ok": False, "error": "write refused — this Local Agent " "was started read-only. Restart with --allow-write to " "permit file writes."} full = _safe(base, job.get("path", "")) content = job.get("content") if content is None: return {"ok": False, "error": "no content"} os.makedirs(os.path.dirname(full) or base, exist_ok=True) with open(full, "w", encoding="utf-8") as f: f.write(str(content)) return {"ok": True, "path": os.path.relpath(full, base), "bytes": len(str(content).encode())} return {"ok": False, "error": f"unknown op: {op}"} def main(): ap = argparse.ArgumentParser(description="Lumen Local Agent") ap.add_argument("--dir", default=os.getcwd(), help="folder to share (default: current directory)") ap.add_argument("--allow-write", action="store_true", help="permit Lumen to WRITE files (default: read-only)") args = ap.parse_args() base = os.path.realpath(os.path.expanduser(args.dir)) if not os.path.isdir(base): print(f"error: {base} is not a directory", file=sys.stderr) sys.exit(1) code = "".join(random.choice("0123456789") for _ in range(6)) mode = "READ + WRITE" if args.allow_write else "READ-ONLY" print("=" * 64) print(" Lumen Local Agent") print("=" * 64) print(f" Shared folder : {base}") print(f" Access mode : {mode}") print(f" Lumen server : {LUMEN}") print() print(f" PAIRING CODE : {code}") print() print(" Sign in at ask.eliteaiempire.com -> Settings -> enter this code.") print(" Lumen can ONLY touch files inside the folder above, and only") print(" while this script is running. Press Ctrl-C to stop & revoke.") print() print(" By running this you consent to local file access as described") print(" at ask.eliteaiempire.com/local-agent. Provided AS IS, no warranty.") print("=" * 64) try: reg = _post("/api/local/agent/register", {"code": code, "dir": base, "allow_write": args.allow_write}) if not reg.get("ok"): print("warning: could not register with Lumen:", reg.get("error", "?"), file=sys.stderr) except Exception as e: print("warning: register failed:", e, file=sys.stderr) print(" Waiting for jobs... (the agent only acts when you ask Lumen " "to use local files)\n") while True: try: job = _post("/api/local/agent/poll", {"code": code}) except Exception: time.sleep(POLL_INTERVAL) continue if job.get("unpaired"): # Lumen forgot the pairing (server restart / expiry) — re-register. try: _post("/api/local/agent/register", {"code": code, "dir": base, "allow_write": args.allow_write}) except Exception: pass time.sleep(POLL_INTERVAL) continue if not job.get("job_id"): time.sleep(POLL_INTERVAL) continue try: result = do_job(job, base, args.allow_write) except Exception as e: result = {"ok": False, "error": str(e)[:200]} print(f" [{time.strftime('%H:%M:%S')}] {job.get('op')} " f"{job.get('path', '')} -> {'ok' if result.get('ok') else 'refused/err'}") try: _post("/api/local/agent/result", {"job_id": job["job_id"], "result": result}) except Exception: pass if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nLumen Local Agent stopped. Local file access revoked.")