From f9309c4dbc42d5d8ac5c0856860514c65c26c413 Mon Sep 17 00:00:00 2001 From: Matthias Guillitte Date: Tue, 18 Aug 2026 17:04:02 +0200 Subject: [PATCH] Authentication and Odata compliance --- cloud/app.py | 501 +++++++++++++++++++++++++++---------------- cloud/odata_query.py | 41 ++++ 2 files changed, 357 insertions(+), 185 deletions(-) create mode 100644 cloud/odata_query.py diff --git a/cloud/app.py b/cloud/app.py index 4df564c..b908e91 100644 --- a/cloud/app.py +++ b/cloud/app.py @@ -1,12 +1,16 @@ import os -import asyncio -import base64 -import uuid -from datetime import datetime, timedelta, timezone import sys import json -from flask import Flask, request, jsonify, current_app -from pymongo import MongoClient +import uuid +import base64 +import asyncio +import fcntl +import atexit +from datetime import datetime, timedelta, timezone +from flask import Flask, Response, request, jsonify, current_app, abort, g +from pymongo import MongoClient, ASCENDING, DESCENDING +from apscheduler.schedulers.background import BackgroundScheduler + from APIs import generate from APIs.mqtt import send_command from APIs.webex import WebexManager @@ -14,8 +18,6 @@ from APIs.shodan import ShodanAuditor from APIs.twilio import send_alert_sms from microwaveCookPlanner import MicrowaveCookPlanner import safety_checker -import pymongo -from apscheduler.schedulers.background import BackgroundScheduler from jobs import job_server_cve_audit, job_client_ip_audit, job_request_telemetry sys.path.insert(0, '..') @@ -35,16 +37,25 @@ MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/") client = MongoClient(MONGO_URI) db = client["microwave_network_db"] +# Database Collections client_collection = db["client_data"] cooking_collection = db["cooking_parameters"] telemetry_collection = db["telemetry_data"] alert_collection = db["alert_data"] -webex_tokens_collection = db["webex_tokens"] # Collection for Webex OAuth tokens -device_ip_collection = db["device_ips"] # Collection for device IPs -cve_audit_results = db["cve_audit_results"] # Collection for CVE audit results +webex_tokens_collection = db["webex_tokens"] +device_ip_collection = db["device_ips"] +cve_audit_results = db["cve_audit_results"] +# Initialize Database Indexes +def init_db_indexes(): + device_ip_collection.create_index([("received_at", DESCENDING)]) + device_ip_collection.create_index([("client_ip", ASCENDING)]) + alert_collection.create_index([("orchestrator_id", ASCENDING), ("created_at", DESCENDING)]) + client_collection.create_index([("devices.device_id", ASCENDING)]) -# Webex Credentials & Configuration from Environment Variables +init_db_indexes() + +# Environment Configurations WEBEX_CLIENT_ID = os.getenv("WEBEX_CLIENT_ID", "YOUR_WEBEX_CLIENT_ID") WEBEX_CLIENT_SECRET = os.getenv("WEBEX_CLIENT_SECRET", "YOUR_WEBEX_CLIENT_SECRET") WEBEX_REDIRECT_URI = os.getenv("WEBEX_REDIRECT_URI", "https://smartwave.matthiasg.dev/oauth/callback") @@ -57,10 +68,9 @@ CAMERA_IMAGE_DIR = "storage/dishPhotos" os.makedirs(CAMERA_IMAGE_DIR, exist_ok=True) # --------------------------------------------------------- -# Classes +# Integrations Setup # --------------------------------------------------------- microwave_cook_planner = MicrowaveCookPlanner() -# Instantiate Webex Manager webex_manager = WebexManager( db_collection=webex_tokens_collection, client_id=WEBEX_CLIENT_ID, @@ -71,60 +81,110 @@ webex_manager = WebexManager( ) shodan_auditor = ShodanAuditor() -# ------------------------------------------------------------- -# Cron Scheduler Setup -# ------------------------------------------------------------- +# --------------------------------------------------------- +# Cron Scheduler (Multi-Worker Safe) +# --------------------------------------------------------- scheduler = BackgroundScheduler(daemon=True) -# Job 1: Server CVE audit every night at 00:00 UTC -scheduler.add_job( - func=job_server_cve_audit, - args=[shodan_auditor, db], - trigger="cron", - hour=0, - minute=0, - id="server_cve_audit_job" -) - -# Job 2: Client IP Shodan audit every 6 hours -scheduler.add_job( - func=job_client_ip_audit, - args=[shodan_auditor, db], - trigger="cron", - hour="*/6", - minute=15, - id="client_ip_audit_job" -) - -# Job 3: Request telemetry runs AFTER the client audit (e.g., every 12 hours) -scheduler.add_job( - func=job_request_telemetry, - args=[send_command], - trigger="cron", - hour="1,13", - minute=0, - id="request_telemetry_job" -) - -scheduler.start() - -# --------------------------------------------------------- -# OAuth Routes -# --------------------------------------------------------- - -@app.route("/oauth/callback") -def oauth_callback(): - """OAuth redirect endpoint that receives the authorization code.""" - code = request.args.get("code") - if not code: - return jsonify({"error": "Missing code parameter"}), 400 - +def start_scheduler_once(): + """Ensures only ONE Gunicorn worker process runs the cron scheduler.""" + lock_file_path = "/tmp/scheduler.lock" + + # Open or create a lock file + lock_file = open(lock_file_path, "wb") + try: - webex_manager.exchange_code(code) - return jsonify({"status": "success", "message": "Webex tokens stored successfully in MongoDB!"}), 200 - except Exception as e: - current_app.logger.exception("Failed to exchange OAuth code") - return jsonify({"error": f"OAuth exchange failed: {str(e)}"}), 500 + # Request a non-blocking exclusive lock + fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + + # Register job schedules + scheduler.add_job( + func=job_server_cve_audit, + args=[shodan_auditor, db], + trigger="cron", + hour=0, + minute=0, + id="server_cve_audit_job", + replace_existing=True + ) + + scheduler.add_job( + func=job_client_ip_audit, + args=[shodan_auditor, db], + trigger="cron", + hour="*/6", + minute=15, + id="client_ip_audit_job", + replace_existing=True + ) + + scheduler.add_job( + func=job_request_telemetry, + args=[send_command], + trigger="cron", + hour="1,13", + minute=0, + id="request_telemetry_job", + replace_existing=True + ) + + scheduler.start() + print(f"[Cron] Scheduler started successfully in Worker PID {os.getpid()}") + + # Ensure lock release on shutdown + def cleanup(): + try: + fcntl.flock(lock_file, fcntl.LOCK_UN) + lock_file.close() + except Exception: + pass + + atexit.register(cleanup) + + except (IOError, OSError): + # Lock acquired by another worker - skip starting scheduler + print(f"[Cron] Worker PID {os.getpid()} skipped scheduler (already running in another worker).") + +# Initialize single-instance scheduler +start_scheduler_once() + +# --------------------------------------------------------- +# Authentication Middleware +# --------------------------------------------------------- +EXEMPT_ROUTES = {'hello_world', 'oauth_callback', 'odata_metadata'} + +@app.before_request +def authenticate_request(): + if request.endpoint in EXEMPT_ROUTES or request.method == 'OPTIONS': + return + + # Extract API Key from Header (X-API-Key or Bearer token) + api_key = request.headers.get("X-API-Key") + if not api_key: + auth_header = request.headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + api_key = auth_header.split(" ")[1] + + if not api_key: + return jsonify({ + "@odata.error": { + "code": "401", + "message": "Unauthorized: Missing API Key header" + } + }), 401 + + # Validate key against 'devices_authentication' collection + device_auth = db.devices_authentication.find_one({"api_key": api_key}) + if not device_auth: + return jsonify({ + "@odata.error": { + "code": "401", + "message": "Unauthorized: Invalid API Key" + } + }), 401 + + # Attach device context (works for both Microwaves and Node-RED) + g.device_id = device_auth.get("device_id") # --------------------------------------------------------- # Application Routes @@ -133,114 +193,130 @@ def oauth_callback(): @app.route("/") def hello_world(): gen = generate(prompt="Say Hello, to the user !") - print(gen) return f"

{gen}

" +@app.route("/oauth/callback") +def oauth_callback(): + code = request.args.get("code") + if not code: + return jsonify({"error": "Missing code parameter"}), 400 + + try: + webex_manager.exchange_code(code) + return jsonify({"status": "success", "message": "Webex tokens stored successfully!"}), 200 + except Exception as e: + current_app.logger.exception("Failed to exchange OAuth code") + return jsonify({"error": "OAuth exchange failed"}), 500 + + @app.route("/cooking-params", methods=["POST"]) async def cooking_params(): try: data = request.get_json() - microwave_id = data.get("microwave_id", "unknown") + if not data or not isinstance(data, dict): + return jsonify({"@odata.error": {"code": "400", "message": "Invalid payload"}}), 400 + + data["microwave_id"] = g.device_id + + forwarded_for = request.headers.get('X-Forwarded-For') + client_ip = forwarded_for.split(',')[0].strip() if forwarded_for else request.remote_addr - client_ip = request.headers.get('X-Forwarded-For', request.remote_addr).split(',')[0].strip() device_ip_collection.insert_one({ - "microwave_id": microwave_id, + "microwave_id": g.device_id, "client_ip": client_ip, "received_at": datetime.now(timezone.utc) }) - - if not data: - return jsonify({"error": "Invalid or missing JSON payload"}), 400 - # Extract user or device parameters height_cm = float(data.get("dish_height", 4.0)) initial_temp_c = float(data.get("ir_initial_temp", 20.0)) microwave_wattage = int(data.get("microwave_wattage", 900)) defrost_mode = bool(data.get("defrost_mode", False)) - # 1. Save Camera Image camera_image_b64 = data.get("camera_image") if not camera_image_b64: - return jsonify({"error": "Missing required field 'camera_image'"}), 400 + return jsonify({"@odata.error": {"code": "400", "message": "Missing camera_image"}}), 400 filename = f"dish_{uuid.uuid4().hex}.jpg" filepath = os.path.join(CAMERA_IMAGE_DIR, filename) - try: - with open(filepath, "wb") as f: - f.write(base64.b64decode(camera_image_b64)) - data["camera_image"] = filepath - except Exception as e: - current_app.logger.exception("Failed to save camera image") - return jsonify({"error": f"Failed to save camera image: {str(e)}"}), 500 + with open(filepath, "wb") as f: + f.write(base64.b64decode(camera_image_b64)) + data["camera_image"] = filepath - # 2. Run Vision Safety Check & Cook Planner Concurrently - try: - if getattr(config, "DEBUG_DANGEROUS_AREA", False): - print(f"[Debug] Running safety check on predefined image") - safety_task = asyncio.to_thread(safety_checker.check_dish_safety, os.path.join(CAMERA_IMAGE_DIR, "dish_0de0ee1dab8949fc8e78796947e24ed3.jpg")) - elif getattr(config, "DEBUG", False): - safety_task = asyncio.to_thread(safety_checker.check_dish_safety, os.path.join("microwaveDish.jpg")) - else: - safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath) - planner_task = asyncio.to_thread( - microwave_cook_planner.generate_plan, - image_path=filepath, - height_cm=height_cm, - initial_temp_c=initial_temp_c, - microwave_wattage=microwave_wattage, - defrost_mode=defrost_mode - ) + # Concurrent AI Execution + safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath) + planner_task = asyncio.to_thread( + microwave_cook_planner.generate_plan, + image_path=filepath, + height_cm=height_cm, + initial_temp_c=initial_temp_c, + microwave_wattage=microwave_wattage, + defrost_mode=defrost_mode + ) - # Execute both concurrently and await results - safety_result, cook_plan = await asyncio.gather(safety_task, planner_task) + safety_result, cook_plan = await asyncio.gather(safety_task, planner_task) - except Exception as e: - current_app.logger.exception("Exception during safety check or cook planning") - return jsonify({"error": f"Task execution failed: {str(e)}"}), 500 - - - # 4. Attach Cooking Plan & safety check and Save to MongoDB data["safety_check"] = safety_result data["analysis_results"] = cook_plan - try: - cooking_collection.insert_one(data) - data.pop("_id", None) - except Exception as e: - return jsonify({"error": f"Database error: {str(e)}"}), 500 + inserted = cooking_collection.insert_one(data) + doc_id = str(inserted.inserted_id) + + # Base OData v4 response wrapper + odata_response = { + "@odata.context": f"{request.host_url.rstrip('/')}/$metadata#CookingParams/$entity", + "@odata.id": f"{request.host_url.rstrip('/')}/cooking-params('{doc_id}')", + "id": doc_id, + "plan": cook_plan + } - # 3. Evaluate Safety Result if not safety_result.get("is_safe", True): - current_app.logger.warning(f"Unsafe dish detected: {safety_result}") - return jsonify({ - "error": "Safety hazard detected in microwave area", + odata_response.update({ "is_safe": False, "warning_message": safety_result.get("warning_message", "Unsafe materials detected."), "detected_hazards": safety_result.get("detected_hazards", []) - }), 200 + }) + return jsonify(odata_response), 200 + + return jsonify(odata_response), 201 - return jsonify(cook_plan), 201 except Exception as e: - current_app.logger.exception("Exception in /cooking-params") - + current_app.logger.exception("Error in /cooking-params") + return jsonify({"@odata.error": {"code": "500", "message": "Internal processing error"}}), 500 @app.route("/alert", methods=["POST"]) def alert(): - data = request.get_json() or {} - - # 1. Inject UTC timestamp + data = request.get_json() + if data is None or not isinstance(data, dict): + return jsonify({ + "@odata.error": { + "code": "400", + "message": "Invalid or missing JSON payload" + } + }), 400 + + # Auto-populate or override orchestrator_id using authenticated device context + orchestrator_id = data.get("orchestrator_id") or getattr(g, "device_id", None) + data["orchestrator_id"] = orchestrator_id data["created_at"] = datetime.now(timezone.utc) - saved_alert = alert_collection.insert_one(data) - - orchestrator_id = data.get("orchestrator_id") - + + try: + saved_alert = alert_collection.insert_one(data) + doc_id = str(saved_alert.inserted_id) + except Exception as e: + current_app.logger.exception("Failed to insert alert into database") + return jsonify({ + "@odata.error": { + "code": "500", + "message": "Database write failed" + } + }), 500 + alert_info = data.get("alert", {}) alert_type = alert_info.get("type", "Unknown") alert_message = alert_info.get("message", "No message provided") - - # --- GET CLIENT METADATA --- + client_name = f"Unknown Client (Orchestrator {orchestrator_id})" if orchestrator_id else "Unknown Client" client_email = None client_phone = None @@ -259,14 +335,12 @@ def alert(): client_name = client_doc.get("client_name", client_name) contact_info = client_doc.get("contact_info", {}) client_email = contact_info.get("email") - client_phone = contact_info.get("phone") # Extract phone number for Twilio + client_phone = contact_info.get("phone") - # --- WEBEX ROOM LOGIC --- webex_room_id = None room_title = f"Support - {client_name}" room_link = None webex_status = "skipped" - # --- TWILIO SMS LOGIC --- sms_sent = False three_minutes_ago = datetime.now(timezone.utc) - timedelta(minutes=3) @@ -276,7 +350,7 @@ def alert(): "webex_room_id": {"$exists": True, "$ne": None}, "created_at": {"$gte": three_minutes_ago}, "_id": {"$ne": saved_alert.inserted_id} - }, sort=[("created_at", pymongo.DESCENDING)]) + }, sort=[("created_at", DESCENDING)]) if recent_alert: webex_room_id = recent_alert["webex_room_id"] @@ -286,15 +360,14 @@ def alert(): room_id=webex_room_id, alert_type=alert_type, alert_message=alert_message, - alert_id=str(saved_alert.inserted_id), + alert_id=doc_id, added_alert_info=True ) webex_status = "reused_room" - room_link = f"https://web.webex.com/spaces/{webex_room_id}" + room_link = recent_alert.get("webex_room_link") or f"https://web.webex.com/spaces/{webex_room_id}" except Exception as e: - current_app.logger.exception("Failed to send follow-up message to existing Webex room") + current_app.logger.exception("Failed to send follow-up message to Webex room") webex_status = f"message_failed: {str(e)}" - else: try: room_details = webex_manager.create_support_room( @@ -302,25 +375,22 @@ def alert(): client_email, alert_type, alert_message, - str(saved_alert.inserted_id) + doc_id ) if isinstance(room_details, dict): webex_room_id = room_details.get("id") room_title = room_details.get("title", room_title) - # Set room_link if your webex_manager returns meetingLink or custom URL room_link = room_details.get("meetingLink") or f"https://web.webex.com/spaces/{webex_room_id}" else: webex_room_id = room_details room_link = f"https://web.webex.com/spaces/{webex_room_id}" webex_status = "created_new_room" - except Exception as e: current_app.logger.exception("Failed to create Webex support room") webex_status = f"creation_failed: {str(e)}" - - # --- SEND SMS --- + if webex_room_id and client_phone and not SAVEUP_TWILIO_API_TOKEN: sms_sent = send_alert_sms( to_phone=client_phone, @@ -328,72 +398,92 @@ def alert(): room_link=room_link, client_email=client_email ) - - # Update document with room and sms details - if webex_room_id: - alert_collection.update_one( - {"_id": saved_alert.inserted_id}, - {"$set": {"webex_room_id": webex_room_id, "webex_room_title": room_title, "webex_room_link": room_link, "sms_sent": sms_sent}} - ) - else: # Just save the sms - alert_collection.update_one( - {"_id": saved_alert.inserted_id}, - {"$set": {"sms_sent": sms_sent}} - ) + update_fields = {"sms_sent": sms_sent} + if webex_room_id: + update_fields.update({ + "webex_room_id": webex_room_id, + "webex_room_title": room_title, + "webex_room_link": room_link + }) + + alert_collection.update_one({"_id": saved_alert.inserted_id}, {"$set": update_fields}) + + # OData v4 Created Response Payload return jsonify({ - "status": "success", - "message": "Alert saved", + "@odata.context": f"{request.host_url.rstrip('/')}/$metadata#Alerts/$entity", + "@odata.id": f"{request.host_url.rstrip('/')}/alert('{doc_id}')", + "id": doc_id, + "status": "success", + "message": "Alert saved", + "orchestrator_id": orchestrator_id, "webex_room_id": webex_room_id, "webex_room_url": room_link, "webex_status": webex_status, "sms_sent": sms_sent - }), 200 - + }), 201 @app.route("/telemetry", methods=["POST"]) def telemetry(): data = request.get_json() - if data is None: - return jsonify({"error": "Invalid or missing JSON payload"}), 400 + return jsonify({"@odata.error": {"code": "400", "message": "Invalid JSON payload"}}), 400 - # If the payload was double-encoded as a string, deserialize it if isinstance(data, str): try: data = json.loads(data) except (json.JSONDecodeError, TypeError): - return jsonify({"error": "String payload could not be parsed as JSON"}), 400 + return jsonify({"@odata.error": {"code": "400", "message": "String payload invalid"}}), 400 - if not isinstance(data, dict): - return jsonify({"error": "Expected a JSON object/dictionary"}), 400 - - # Stamp UTC timestamp for Node-RED queries + data["device_id"] = g.device_id data["received_at"] = datetime.now(timezone.utc).isoformat() try: - telemetry_collection.insert_one(data) - return jsonify({"status": "success", "message": "Telemetry saved"}), 200 - - except Exception as e: - return jsonify({"error": f"Database error: {str(e)}"}), 500 + inserted = telemetry_collection.insert_one(data) + doc_id = str(inserted.inserted_id) -# ------------------------------------------------------------- -# Server CVE & SSL Audit Endpoint -# ------------------------------------------------------------- -@app.route('/api/security/cve-audit', methods=['GET']) + # OData v4 Created Response Envelope + return jsonify({ + "@odata.context": f"{request.host_url.rstrip('/')}/$metadata#Telemetry/$entity", + "@odata.id": f"{request.host_url.rstrip('/')}/telemetry('{doc_id}')", + "id": doc_id, + "status": "success", + "message": "Telemetry saved", + "device_id": g.device_id, + "received_at": data["received_at"] + }), 201 + + except Exception as e: + current_app.logger.exception("Database write failed") + return jsonify({"@odata.error": {"code": "500", "message": "Database write failed"}}), 500 + + +@app.route('/api/security/cve-audit', methods=['POST', 'GET']) def run_cve_audit(): - """Runs a vulnerability and SSL audit against this server.""" + """Protected endpoint triggered by Node-RED using REST API + API Key.""" audit_results = shodan_auditor.audit_server_vulnerabilities() - cve_audit_results.insert_one({ + inserted = cve_audit_results.insert_one({ + "triggered_by": g.device_id, "audit_results": audit_results, "timestamp": datetime.now(timezone.utc) }) + doc_id = str(inserted.inserted_id) - status_code = 200 if audit_results["status"] == "PASS" else 409 - return jsonify(audit_results), status_code + status_code = 200 if audit_results.get("status") == "PASS" else 409 + + return jsonify({ + "@odata.context": f"{request.host_url.rstrip('/')}/$metadata#CveAudit/$entity", + "@odata.id": f"{request.host_url.rstrip('/')}/api/security/cve-audit('{doc_id}')", + "id": doc_id, + "status": audit_results.get("status"), + "audit_results": audit_results, + "timestamp": datetime.now(timezone.utc).isoformat() + }), status_code +# --------------------------------------------------------- +# Debug Endpoints (Protected) +# --------------------------------------------------------- @app.route("/debug/telemetryrequest", methods=["GET"]) def debug_telemetryrequest(): @@ -413,7 +503,8 @@ def debug_telemetryrequest(): }), 200 except Exception as e: return jsonify({"error": f"Failed to publish MQTT command: {str(e)}"}), 500 - + + @app.route('/debug/run-job/', methods=['POST', 'GET']) def trigger_job_manually(job_id): """Manually triggers any scheduled job immediately by its ID.""" @@ -424,7 +515,6 @@ def trigger_job_manually(job_id): "available_jobs": [j.id for j in scheduler.get_jobs()] }), 404 - # Execute the underlying function with its configured arguments try: job.func(*job.args) return jsonify({ @@ -436,10 +526,51 @@ def trigger_job_manually(job_id): "status": "error", "message": f"Job execution failed: {str(e)}" }), 500 - -@app.route("/debug", methods=["GET"]) -def debug(): - return safety_checker.check_dish_safety(os.path.join(CAMERA_IMAGE_DIR, "dish_0de0ee1dab8949fc8e78796947e24ed3.jpg")) + +# --------------------------------------------------------- +# OData Metadata Definition +# --------------------------------------------------------- +@app.route("/$metadata", methods=["GET"]) +def odata_metadata(): + """Provides mandatory OData schema definition for validation.""" + xml_metadata = """ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +""" + return Response(xml_metadata, mimetype="application/xml") if __name__ == "__main__": diff --git a/cloud/odata_query.py b/cloud/odata_query.py new file mode 100644 index 0000000..92d66e9 --- /dev/null +++ b/cloud/odata_query.py @@ -0,0 +1,41 @@ +def parse_odata_query(request_args): + """ + Translates standard OData query parameters ($top, $skip, $orderby, $filter) + into MongoDB query parameters. + """ + limit = int(request_args.get('$top', 50)) + skip = int(request_args.get('$skip', 0)) + + # Parse $orderby (e.g., $orderby=received_at desc) + sort = [] + orderby_str = request_args.get('$orderby') + if orderby_str: + parts = orderby_str.split() + field = parts[0] + direction = -1 if len(parts) > 1 and parts[1].lower() == 'desc' else 1 + sort.append((field, direction)) + + # Parse basic $filter (e.g., $filter=device_id eq '1') + mongo_filter = {} + filter_str = request_args.get('$filter') + if filter_str and ' eq ' in filter_str: + field, val = filter_str.split(' eq ') + field = field.strip() + val = val.strip().strip("'").strip('"') + + # Cast data types + if val.isdigit(): + val = int(val) + elif val.lower() == 'true': + val = True + elif val.lower() == 'false': + val = False + + mongo_filter[field] = val + + return { + "filter": mongo_filter, + "limit": limit, + "skip": skip, + "sort": sort or [("_id", -1)] + } \ No newline at end of file