otb-cloud/app/auth/utils.py

import hashlib
import hmac
import re
import time
from pathlib import Path

from flask import current_app

from app.db import get_db

SLUG_RE = re.compile(r"[^a-z0-9]+")

def make_signature(email: str, ts: str, portal_user_id: str, secret: str) -> str:
    payload = f"{portal_user_id}|{email}|{ts}".encode("utf-8")
    return hmac.new(secret.encode("utf-8"), payload, hashlib.sha256).hexdigest()

def is_valid_signature(email: str, ts: str, portal_user_id: str, sig: str) -> bool:
    secret = current_app.config["OTB_PORTAL_SHARED_SECRET"]
    expected = make_signature(email=email, ts=ts, portal_user_id=portal_user_id, secret=secret)
    return hmac.compare_digest(expected, sig)

def is_valid_timestamp(ts: str) -> bool:
    try:
        ts_int = int(ts)
    except ValueError:
        return False
    skew = current_app.config["OTB_PORTAL_ALLOWED_SKEW_SECONDS"]
    return abs(int(time.time()) - ts_int) <= skew

def slugify_email(email: str) -> str:
    local = email.split("@", 1)[0].lower().strip()
    slug = SLUG_RE.sub("-", local).strip("-")
    return slug or "tenant"

def ensure_user_tenant_and_devices(email: str, portal_user_id: int):
    db = get_db()

    with db.cursor() as cur:
        cur.execute(
            """
            SELECT id, portal_user_id, email, display_name
            FROM users
            WHERE email = %s
            """,
            (email,),
        )
        user = cur.fetchone()

        if user is None:
            cur.execute(
                """
                INSERT INTO users (portal_user_id, email, display_name, is_active)
                VALUES (%s, %s, %s, 1)
                """,
                (portal_user_id, email, email),
            )
            user_id = cur.lastrowid
        else:
            user_id = user["id"]
            cur.execute(
                """
                UPDATE users
                SET portal_user_id = %s, last_login_at = NOW()
                WHERE id = %s
                """,
                (portal_user_id, user_id),
            )

        cur.execute(
            """
            SELECT id, slug, storage_root
            FROM tenants
            WHERE owner_user_id = %s
            """,
            (user_id,),
        )
        tenant = cur.fetchone()

        if tenant is None:
            base_slug = slugify_email(email)
            slug = base_slug
            suffix = 1

            while True:
                cur.execute("SELECT id FROM tenants WHERE slug = %s", (slug,))
                existing = cur.fetchone()
                if existing is None:
                    break
                suffix += 1
                slug = f"{base_slug}-{suffix}"

            storage_root = f"{current_app.config['STORAGE_ROOT']}/tenants/{slug}"
            cur.execute(
                """
                INSERT INTO tenants (owner_user_id, slug, storage_root, service_status, retention_mode)
                VALUES (%s, %s, %s, 'active', 'standard')
                """,
                (user_id, slug, storage_root),
            )
            tenant_id = cur.lastrowid
        else:
            tenant_id = tenant["id"]
            slug = tenant["slug"]
            storage_root = tenant["storage_root"]

        for device_name in current_app.config["DEFAULT_DEVICE_NAMES"]:
            relative_path = f"devices/{device_name}"
            cur.execute(
                """
                INSERT IGNORE INTO devices (tenant_id, device_name, device_type, relative_path, is_active)
                VALUES (%s, %s, %s, %s, 1)
                """,
                (tenant_id, device_name, device_name, relative_path),
            )

        db.commit()

    create_tenant_directories(slug, current_app.config["DEFAULT_DEVICE_NAMES"])

    return {
        "user_id": user_id,
        "tenant_id": tenant_id,
        "tenant_slug": slug,
        "storage_root": storage_root,
        "email": email,
    }

def create_tenant_directories(tenant_slug: str, device_names: list[str]):
    tenant_root = Path(current_app.config["STORAGE_ROOT"]) / "tenants" / tenant_slug
    (tenant_root / "logs").mkdir(parents=True, exist_ok=True)
    (tenant_root / "support").mkdir(parents=True, exist_ok=True)

    for device_name in device_names:
        for subdir in ["originals", "derived", "exports", "deleted", "tmp"]:
            (tenant_root / "devices" / device_name / subdir).mkdir(parents=True, exist_ok=True)