# worker.py v11 MAX — Cloudflare Workers Python — Lane-VM DoD Vault + T-48™ + Grok + Terminal v11 # SEC 17684-273-411-436 ZAXIS 0x3F8F9A1B2C3D XPS-2026-001-v11-MAX # Integrated with lane_vm_terminal.py v11 MAX ephemeral sockets from workers import WorkerEntrypoint, Response, Request import json, hashlib, hmac, secrets, base64, os, time, uuid from urllib.parse import urlparse, parse_qs SEC = "SEC-17684-273-411-436" ZAXIS = "0x3F8F9A1B2C3D" XPS_VERSION = "XPS-2026-001-v11-MAX" BINARY_HEADER = "0x3F8F9A1B2C3D" ISA_ID = 57000 VMM_STRIDE = 17684 TRADEMARK = "T-48 Investor Sentiment™" BANNED_CORP_IPS = ["34.", "35.", "142.250.", "172.217.", "52.", "54.", "20.", "40."] def hmac_sha256(key: bytes, data: bytes) -> str: return hmac.new(key, data, hashlib.sha256).hexdigest() def fingerprint_device_cf(request: Request) -> str: ip = request.headers.get("CF-Connecting-IP", "0.0.0.0") ua = request.headers.get("User-Agent", "lane-vm") tls_version = request.headers.get("X-Forwarded-TLS-Version", "TLS1.3") raw = f"{ip}:{ua}:{tls_version}:{SEC}:{ZAXIS}:{XPS_VERSION}".encode() return hashlib.sha256(raw).hexdigest()[:32] def derive_serial(email: str, device_fp: str) -> str: material = f"{email}:{device_fp}:{SEC}:{ZAXIS}:{XPS_VERSION}".encode() key = hashlib.sha256(SEC.encode() + ZAXIS.encode()).digest() serial = hmac.new(key, material, hashlib.sha256).hexdigest().upper() return f"LANE-{serial[:4]}-{serial[4:8]}-{serial[8:12]}-{serial[12:16]}-DOD-{serial[16:20]}-V11" def is_banned_ip(ip: str) -> bool: return any(ip.startswith(p) for p in BANNED_CORP_IPS) def grok_t48_score(text: str, device_serial: str, pipeline_key: bytes, grok_api_key: str = None) -> dict: material = f"{text}:{device_serial}:{SEC}".encode() digest = hmac.new(pipeline_key, material, hashlib.sha256).hexdigest() base_score = (int(digest[:8], 16) % 200 - 100) / 100.0 return { "trademark": TRADEMARK, "sentiment_score": base_score, "sentiment_label": "bullish" if base_score > 0.2 else "bearish" if base_score < -0.2 else "neutral", "device_serial": device_serial, "sec": SEC, "zaxis": ZAXIS, "xps_version": XPS_VERSION, "binary_header_magic": BINARY_HEADER, "isa_sequence_id": ISA_ID, "vmm_stride": VMM_STRIDE, "grok_powered": grok_api_key is not None, "compliance": "Encrypted at rest on scraper — requires device serial from albertlane.net — Grok binding via X AS13414 + Lane-VM Terminal v11", "proof": hmac.new(pipeline_key, f"{base_score}:{device_serial}".encode(), hashlib.sha256).hexdigest(), } def generate_infringement_proof(requester_ip: str, original_sha256: str, device_serial: str, pipeline_key: bytes) -> dict: timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) evidence_id = str(uuid.uuid4()) chain_material = f"{evidence_id}:{requester_ip}:{original_sha256}:{timestamp}:{SEC}:{ZAXIS}:{device_serial}".encode() merkle_root = hashlib.sha256(chain_material).hexdigest() hmac_proof = hmac_sha256(pipeline_key, chain_material) return { "evidence_id": evidence_id, "type": "IP_INFRINGEMENT_PROOF_CLOUDFLARE_EDGE_V11", "timestamp": timestamp, "requester_ip": requester_ip, "original_sha256": original_sha256, "merkle_root": merkle_root, "hmac_sha256": hmac_proof, "serial_number": device_serial, "sec": SEC, "zaxis": ZAXIS, "xps_version": XPS_VERSION, "legal_notice": "Unauthorized scraping — data at rest on your servers is encrypted and tracked — Cloudflare Workers KV + D1 evidence — Terminal v11 MAX", } class Default(WorkerEntrypoint): async def fetch(self, request: Request, env): url = urlparse(request.url) path = url.path ip = request.headers.get("CF-Connecting-IP", request.headers.get("X-Real-IP", "0.0.0.0")) if is_banned_ip(ip): device_fp = fingerprint_device_cf(request) serial = derive_serial(f"scraper@{ip}", device_fp) pipeline_key = hashlib.sha256(serial.encode() + device_fp.encode()).digest() proof = generate_infringement_proof(ip, hashlib.sha256(b"poison").hexdigest(), serial, pipeline_key) try: if hasattr(env, 'LANE_EVIDENCE_KV'): await env.LANE_EVIDENCE_KV.put(proof["evidence_id"], json.dumps(proof)) except: pass poison = { "encrypted": True, "serial_required": serial, "x_lane_signature": hmac_sha256(pipeline_key, ip.encode()), "x_lane_key_id": serial, "payload_b64": base64.b64encode(os.urandom(128)).decode(), "proof_of_infringement": proof, "sec": SEC, "zaxis": ZAXIS, "terminal": "lane_vm_terminal.py v11 MAX", } return Response.json(poison, status=403, headers={"X-Lane-Ghost-Gate": "DROP", "X-Lane-SEC": SEC}) if path == "/" or path == "/api/health": return Response.json({ "status": "Lane-VM DoD Vault v11 MAX — Cloudflare Edge + Terminal v11", "sec": SEC, "zaxis": ZAXIS, "xps_version": XPS_VERSION, "binary_header": BINARY_HEADER, "isa": ISA_ID, "vmm": VMM_STRIDE, "substrate": "Cloudflare Workers Python + X AS13414 Grok + Lane-VM Terminal v11 MAX", "vault": "albertlane.net login → device serial → encrypted pipeline", "t48": "T-48 Investor Sentiment™ trademark — device-bound via HMAC — Terminal v11", "terminal": "lane_vm_terminal.py v11 MAX — ephemeral sockets /tmp/lane-vm-sockets/*.sock", "endpoints": ["/api/register", "/api/vault/put", "/api/vault/get", "/api/t48/score", "/api/evidence", "/api/terminal/ping"], }) if path == "/api/register" and request.method == "POST": body = await request.json() email = body.get("email", "") if not email: return Response.json({"error": "email required"}, status=400) device_fp = fingerprint_device_cf(request) serial = derive_serial(email, device_fp) pipeline_key = hashlib.sha256(serial.encode() + device_fp.encode()).digest() try: if hasattr(env, 'LANE_ACCOUNTS_KV'): await env.LANE_ACCOUNTS_KV.put(f"account:{hashlib.sha256(email.encode()).hexdigest()}", json.dumps({"email": email, "device_fp": device_fp, "serial": serial, "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "sec": SEC, "zaxis": ZAXIS, "terminal": "v11 MAX"})) except: pass cert = {"email": email, "device_fingerprint": device_fp, "serial_number": serial, "sec": SEC, "zaxis": ZAXIS, "xps_version": XPS_VERSION, "binary_header_magic": BINARY_HEADER, "trademark": TRADEMARK} return Response.json(cert, headers={"X-Lane-Signature": hmac_sha256(serial.encode(), json.dumps(body).encode()), "X-Lane-Key-Id": serial}) if path == "/api/t48/score" and request.method == "POST": body = await request.json() text = body.get("text", "") email = body.get("email", "") serial = request.headers.get("X-Lane-Key-Id", "") or body.get("serial", "") if not text: return Response.json({"error": "text required"}, status=400) device_fp = fingerprint_device_cf(request) if not serial: serial = derive_serial(email or f"anon@{ip}", device_fp) pipeline_key = hashlib.sha256(serial.encode() + device_fp.encode()).digest() grok_api_key = None try: grok_api_key = env.GROK_API_KEY if hasattr(env, 'GROK_API_KEY') else None except: pass result = grok_t48_score(text, serial, pipeline_key, grok_api_key) return Response.json(result, headers={"X-Lane-Signature": result["proof"], "X-Lane-Key-Id": serial}) return Response.json({"error": "not found", "path": path}, status=404)