"""Authentication and trial gating module for Orionus demo. Handles email registration, verification code generation/sending, one-use-per-email enforcement, and session time limits. """ from __future__ import annotations import json import os import random import smtplib import string import time from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from pathlib import Path # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- USED_EMAILS_FILE = Path(__file__).resolve().parent / "used_emails.json" TRIAL_DURATION_SEC = 60 # 1-minute demo session VIDEO_MAX_DURATION_SEC = 60 # 1-minute max for uploaded video LIVE_SESSION_MAX_SEC = 60 # 1-minute max for live camera session CODE_LENGTH = 6 CODE_EXPIRY_SEC = 300 # code valid for 5 minutes # SMTP configuration from environment SMTP_HOST = os.environ.get("SMTP_HOST", "smtp-relay.brevo.com") SMTP_PORT = int(os.environ.get("SMTP_PORT", "587")) SMTP_USER = os.environ.get("SMTP_USER", "a7fb01001@smtp-brevo.com") SMTP_PASS = os.environ.get("SMTP_PASS", "xsmtpsib-95d4419e874210d694181e185f1a19f0a13d14f09176441701adfbcc2ceab161-gfhZ4hNblHSUa6CM") SMTP_FROM = os.environ.get("SMTP_FROM", "info@caitcore.com") BREVO_API_KEY = os.environ.get("BREVO_API_KEY", "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") # --------------------------------------------------------------------------- # Email storage (simple JSON file) # --------------------------------------------------------------------------- def _load_used_emails() -> dict: """Load the used-emails registry from disk.""" if not USED_EMAILS_FILE.exists(): return {} try: with open(USED_EMAILS_FILE, "r", encoding="utf-8") as fh: return json.load(fh) except (json.JSONDecodeError, OSError): return {} def _save_used_emails(data: dict) -> None: """Persist the used-emails registry to disk.""" with open(USED_EMAILS_FILE, "w", encoding="utf-8") as fh: json.dump(data, fh, indent=2) def is_email_used(email: str) -> bool: """Return True if this email has already been used for a trial.""" data = _load_used_emails() return email.lower().strip() in data def mark_email_used(email: str) -> None: """Record that this email has completed its trial.""" data = _load_used_emails() data[email.lower().strip()] = { "timestamp": time.time(), } _save_used_emails(data) # --------------------------------------------------------------------------- # Verification code # --------------------------------------------------------------------------- def generate_code() -> str: """Generate a random 6-digit numeric verification code.""" return "".join(random.choices(string.digits, k=CODE_LENGTH)) def smtp_configured() -> bool: """Return True if SMTP env vars are set.""" print(f"[Auth] SMTP check: HOST={SMTP_HOST}, USER={SMTP_USER[:10]}..., PASS={'set' if SMTP_PASS else 'empty'}, FROM={SMTP_FROM}") return bool(SMTP_HOST and SMTP_USER and SMTP_PASS) def _build_html_email(code: str) -> str: """Build branded HTML email for Orionus.""" return f"""

Orionus

Real-Time Emotion Recognition

Your demo access code:

{code}

One-time code • Session lasts {TRIAL_DURATION_SEC}s • No data stored

""" def _send_via_http(to_email: str, code: str) -> bool: """Send email via Brevo HTTP API.""" import urllib.request import urllib.error html_body = _build_html_email(code) payload = json.dumps({ "sender": {"name": "CAIT", "email": SMTP_FROM}, "to": [{"email": to_email}], "subject": f"Orionus Demo Access Code: {code}", "htmlContent": html_body, }) req = urllib.request.Request( "https://api.brevo.com/v3/smtp/email", data=payload.encode("utf-8"), headers={ "accept": "application/json", "content-type": "application/json", "api-key": BREVO_API_KEY, }, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: print(f"[Auth] HTTP API response: {resp.status} {resp.read().decode()}") return True except urllib.error.HTTPError as e: body = e.read().decode() if e.fp else "" print(f"[Auth] HTTP API error: {e.code} {body}") return False except Exception as e: print(f"[Auth] HTTP API exception: {e}") return False def _send_via_smtp(to_email: str, code: str) -> bool: """Send email via SMTP (fallback).""" html_body = _build_html_email(code) text_body = ( f"Your Orionus demo verification code is: {code}\n\n" f"This code expires in {CODE_EXPIRY_SEC // 60} minutes.\n\n" f"-- Orionus AI | CAIT Core" ) msg = MIMEMultipart("alternative") msg["Subject"] = f"Orionus Demo Access Code: {code}" msg["From"] = SMTP_FROM msg["To"] = to_email msg.attach(MIMEText(text_body, "plain")) msg.attach(MIMEText(html_body, "html")) with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10) as srv: srv.ehlo() srv.starttls() srv.ehlo() srv.login(SMTP_USER, SMTP_PASS) srv.sendmail(SMTP_FROM, to_email, msg.as_string()) return True def send_verification_email(email: str, code: str) -> bool: """Send verification code — tries HTTP API first, falls back to SMTP.""" if not smtp_configured(): return False try: if _send_via_http(email, code): return True print("[Auth] HTTP API failed, trying SMTP...") except Exception as e: print(f"[Auth] HTTP API exception: {e}, trying SMTP...") try: return _send_via_smtp(email, code) except Exception as e: print(f"[Auth] SMTP send FAILED: {e}") return False # --------------------------------------------------------------------------- # Session timer helpers # --------------------------------------------------------------------------- def session_expired(start_time: float) -> bool: """Return True if the trial session has exceeded its time limit.""" return (time.time() - start_time) >= TRIAL_DURATION_SEC def remaining_seconds(start_time: float) -> int: """Return seconds remaining in the trial, clamped to >= 0.""" left = TRIAL_DURATION_SEC - (time.time() - start_time) return max(0, int(left))