import os import json from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from pydantic_settings import BaseSettings from typing import Optional def _resolve_db_url(env_url: str) -> str: """Read DB config from wizard config volume if available, else use env.""" config_dir = os.getenv("CONFIG_DIR", "/config") config_file = os.getenv("CONFIG_FILE", "harborforge.json") config_path = os.path.join(config_dir, config_file) if os.path.exists(config_path): try: with open(config_path, "r") as f: cfg = json.load(f) db_cfg = cfg.get("database") if db_cfg: host = db_cfg.get("host", "mysql") port = db_cfg.get("port", 3306) user = db_cfg.get("user", "harborforge") password = db_cfg.get("password", "harborforge_pass") database = db_cfg.get("database", "harborforge") return f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}" except Exception: pass return env_url class Settings(BaseSettings): DATABASE_URL: str = "mysql+pymysql://harborforge:harborforge_pass@mysql:3306/harborforge" SECRET_KEY: str = "change-me-in-production" LOG_LEVEL: str = "INFO" ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 # --- OIDC (generic, OpenID Connect discovery) --- OIDC_ENABLED: bool = False OIDC_ISSUER: str = "" # e.g. https://idp.example.com (we use {issuer}/.well-known/openid-configuration) OIDC_CLIENT_ID: str = "" OIDC_CLIENT_SECRET: str = "" OIDC_REDIRECT_URI: str = "" # backend callback, e.g. https://hf-api.example.com/auth/oidc/callback OIDC_SCOPES: str = "openid email profile" OIDC_POST_LOGIN_REDIRECT: str = "" # frontend URL to return to (token in fragment). Falls back to "/" OIDC_ADMIN_ROLE: str = "admin" # OIDC role name that bootstraps the unbound hf admin (OIDC-only) # When true: no password login at all. Password login endpoint rejects, # user creation ignores any password (passwordless user that can only use # API keys / OIDC), and the frontend hides all password UI. HARBORFORGE_OIDC_ONLY: bool = False class Config: env_file = ".env" settings = Settings() # Fail fast on a weak/default JWT signing key (prevents token forgery). _WEAK_SECRETS = { "change-me-in-production", "change_me_in_production", "change-me-use-openssl-rand-hex-32", "secret", "changeme", "", } if settings.SECRET_KEY in _WEAK_SECRETS or len(settings.SECRET_KEY) < 32: raise RuntimeError( "Insecure SECRET_KEY: set a strong random value " "(e.g. `openssl rand -hex 32`) via the SECRET_KEY env var. " "Refusing to start with a default/short key." ) # Resolve DB URL: wizard config volume > env > default _db_url = _resolve_db_url(settings.DATABASE_URL) engine = create_engine(_db_url, pool_pre_ping=True) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def get_db(): db = SessionLocal() try: yield db finally: db.close()