Spaces:
Running
Running
| """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""" | |
| <html><body style="font-family: -apple-system, sans-serif; background: #0A0A1A; color: #C0D0C0; padding: 40px;"> | |
| <div style="max-width: 520px; margin: 0 auto; background: linear-gradient(135deg, #0D2818, #162B1E); | |
| border: 1px solid rgba(76,175,80,0.2); border-radius: 16px; padding: 40px;"> | |
| <div style="text-align: center; margin-bottom: 24px;"> | |
| <img src="https://caitcore.com/images/orionus-logo.png" alt="Orionus" style="width: 80px; height: 80px; border-radius: 16px;" /> | |
| </div> | |
| <h1 style="color: #81C784; font-size: 26px; text-align: center; margin: 0 0 8px 0;"> | |
| Orionus | |
| </h1> | |
| <p style="text-align: center; color: #6B9B6E; font-size: 13px; margin: 0 0 24px 0;">Real-Time Emotion Recognition</p> | |
| <p style="text-align: center; color: #8FAF8F; font-size: 14px;">Your demo access code:</p> | |
| <div style="text-align: center; font-size: 36px; font-weight: 800; letter-spacing: 8px; | |
| color: #66BB6A; padding: 20px; background: rgba(76,175,80,0.08); | |
| border-radius: 12px; margin: 16px 0;"> | |
| {code} | |
| </div> | |
| <p style="text-align: center; color: #6B9B6E; font-size: 12px; margin-top: 24px;"> | |
| One-time code • Session lasts {TRIAL_DURATION_SEC}s • No data stored | |
| </p> | |
| <hr style="border: none; border-top: 1px solid rgba(76,175,80,0.08); margin: 28px 0 16px;" /> | |
| <div style="text-align: center;"> | |
| <img src="https://caitcore.com/images/cait-logo.png" alt="CAIT" style="width: 32px; height: 32px; vertical-align: middle; border-radius: 6px;" /> | |
| <span style="color: #556677; font-size: 11px; vertical-align: middle; margin-left: 8px;">Powered by CAIT — Computational & AI Technologies</span> | |
| </div> | |
| </div> | |
| </body></html> | |
| """ | |
| 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)) | |