# File: /Users/bpa/Desktop/ForesixaQuant/mobile_gateway/backend/consigliere_server.py # Configuration: Consigliere Mobile Gateway Core Enterprise v7 (Protocol Version Matched) import argparse import asyncio import hmac import json import sqlite3 from pathlib import Path from aiohttp import web, WSMsgType MAX_REQUEST_BYTES = 16384 class HardenedConsigliereGateway: def __init__(self, root_path, token, origin): self.root_path = Path(root_path) self.token = token.encode("utf-8") self.allowed_origin = origin self.db_path = self.root_path / "trading_live.db" self.access_json_path = self.root_path / "mobile_access.json" self.connected_sockets = set() def verify_token_constant_time(self, inbound_token_str): return hmac.compare_digest(inbound_token_str.encode("utf-8"), self.token) def fetch_aligned_ledger_rows(self): db_uri = f"file:{self.db_path}?mode=ro&uri=true" try: conn = sqlite3.connect(db_uri, uri=True) cursor = conn.cursor() cursor.execute("SELECT id, timestamp, symbol, side, qty, price, allocation_spent, status, alpaca_order_id FROM corporate_trade_ledger ORDER BY id DESC LIMIT 5") trades = cursor.fetchall() conn.close() return [ {"id": t, "timestamp": t, "symbol": t, "side": t, "qty": str(t), "price": str(t), "allocation_spent": str(t), "status": t, "alpaca_order_id": t} for t in trades ] except Exception as e: print(f"⚠️ [DATABASE] Read Exception: {e}") return [] async def handle_snapshot_lookup(self, request): auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): return web.json_response({"error": "Unauthorized"}, status=401) inbound_token = auth_header.split(" ").strip() if not self.verify_token_constant_time(inbound_token): return web.json_response({"error": "Unauthorized"}, status=401) try: trade_matrices = self.fetch_aligned_ledger_rows() return web.json_response({"status": "SNAPSHOT_PROCESSED_SUCCESS", "active_ledgers": trade_matrices}) except Exception as e: return web.json_response({"error": "Database error", "details": str(e)}, status=500) async def handle_websocket_stream(self, request): ws = web.WebSocketResponse(autoclose=True, autoping=True, heartbeat=15.0) await ws.prepare(request) self.connected_sockets.add(ws) print("📶 [WEBSOCKET CONNECTED] Secure phone data stream channel opened.") try: async for msg in ws: if msg.type == WSMsgType.TEXT: data = json.loads(msg.data) if data.get("type") == "authenticate" or data.get("type") == "auth": inbound_token = data.get("token", "") if self.verify_token_constant_time(inbound_token): await ws.send_str(json.dumps({"type": "authenticated", "status": "success"})) trade_matrices = self.fetch_aligned_ledger_rows() await ws.send_str(json.dumps({ "type": "snapshot", "data": { "portfolio": {"status": "ACTIVE", "balance": "100000.00"}, "research": {"signals": []}, "ledger": trade_matrices } })) if self.access_json_path.exists(): with open(self.access_json_path, "r") as f: access_data = json.load(f) access_data["pairing_active"] = True with open(self.access_json_path, "w") as f: json.dump(access_data, f, indent=2) else: await ws.send_str(json.dumps({"type": "auth_result", "status": "failed"})) await ws.close() elif data.get("type") == "ping": await ws.send_str(json.dumps({"type": "pong"})) finally: if ws in self.connected_sockets: self.connected_sockets.remove(ws) if self.access_json_path.exists(): with open(self.access_json_path, "r") as f: access_data = json.load(f) access_data["pairing_active"] = False with open(self.access_json_path, "w") as f: json.dump(access_data, f, indent=2) print("🛑 [WEBSOCKET DISCONNECTED] Secure phone data stream channel closed.") return ws async def handle_root_health(self, request): try: with open(self.access_json_path, "r") as f: data = json.load(f) pairing_status = data.get("pairing_active", False) except: pairing_status = False # Return the explicit structural layout metadata keys expected by the application return web.json_response({ "gateway_status": "ONLINE", "pairing_active": pairing_status, "version": "1.0.0", "service": "qzotic_gateway", "runtime_environment": "Python_3.11_Sovereign_Venv" }) def setup_routing_matrix(gateway_instance): app = web.Application() async def cors_middleware(app, handler): async def middleware_handler(request): if request.method == "OPTIONS": response = web.Response(status=204) else: response = await handler(request) response.headers["Access-Control-Allow-Origin"] = gateway_instance.allowed_origin response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS" response.headers["Access-Control-Allow-Headers"] = "Authorization, Content-Type" return response return middleware_handler app.middlewares.append(cors_middleware) app.router.add_get("/", gateway_instance.handle_root_health) app.router.add_get("/api/snapshot", gateway_instance.handle_snapshot_lookup) app.router.add_get("/api/stream", gateway_instance.handle_websocket_stream) return app def main(): p = argparse.ArgumentParser() p.add_argument('--root', type=Path, default=Path('/Users/bpa/Desktop/ForesixaQuant')) p.add_argument('--host', default='127.0.0.1') p.add_argument('--port', type=int, default=8765) p.add_argument('--origin', required=True) args = p.parse_args() with open(args.root / "mobile_access.json", "r") as f: config = json.load(f) active_token = config.get("token") gateway = HardenedConsigliereGateway(args.root, active_token, args.origin) app = setup_routing_matrix(gateway) print(f"\n============================================================") print(f" CONSIGLIERE MOBILE APP GATEWAY: INITIALIZED SUCCESSFUL ") print(f"============================================================") print(f" [BOOT] Listening for incoming secure syncs at: http://{args.host}:{args.port}") print(f" [DATA] Multi-dimensional mapping target: {args.root} (Protocol Version v1.0.0 Aligned)") print(f"============================================================\n") runner = web.AppRunner(app) asyncio.get_event_loop().run_until_complete(runner.setup()) site = web.TCPSite(runner, args.host, args.port) asyncio.get_event_loop().run_until_complete(site.start()) print(" [BOOT] Sockets Bound Successfully. Origin Flag Enforcement Live.") asyncio.get_event_loop().run_forever() if __name__ == "__main__": main()