gtiandClaude Opus 5.5
64d3fc56e2
Collaborators logged into GTI Gerenciador open GTI Fiscal without a second password. The panel signs a 60 s JWT (Ed25519; the Hub holds only the public key); the Hub checks signature, iss/aud/exp/nbf, maximum lifetime, nonce against a one-time state cookie set when the flow starts at GET /login (login-CSRF protection) and a single-use jti (sso_used_assertions). Collaborators are provisioned in admin_users with auth_source=GTI_GERENCIADOR and no password; a CHECK constraint makes a passwordless local account or a federated account with a password impossible (migration 0021). Sessions last FEDERATED_SESSION_SECONDS (1 h) and are renewed silently through the panel. The local admin stays as break-glass at /login/local, audited. Logout ends only the Hub session. Every SSO success/rejection (stable reason code), logout and local login is audited with the real client IP, never the token. verify_password now fails closed on a missing hash instead of raising. Visual identity: gti-theme.css is a verbatim copy of the panel's design system (tokens, sidebar, light/dark theme) plus its tema.js, sidebar.js and logo; control-panel.css keeps only Hub components, all colours from the panel tokens. Page structure and features unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
"""Separate human authentication backed by hashed passwords and sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import HTTPException, Request
|
|
from sqlalchemy import select
|
|
|
|
from .config import settings
|
|
from .models import AdminUser, WebSession
|
|
|
|
|
|
def _digest(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = os.urandom(16)
|
|
encoded = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 240_000)
|
|
return "pbkdf2_sha256$240000$%s$%s" % (base64.urlsafe_b64encode(salt).decode(), base64.urlsafe_b64encode(encoded).decode())
|
|
|
|
|
|
def verify_password(password: str, stored: str | None) -> bool:
|
|
# Federated (SSO) accounts have no password: always refuse, never raise.
|
|
if not isinstance(stored, str) or not isinstance(password, str):
|
|
return False
|
|
try:
|
|
algorithm, rounds, salt, digest = stored.split("$", 3)
|
|
candidate = hashlib.pbkdf2_hmac(algorithm.removeprefix("pbkdf2_"), password.encode(), base64.urlsafe_b64decode(salt), int(rounds))
|
|
return hmac.compare_digest(base64.urlsafe_b64encode(candidate).decode(), digest)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def create_admin(session, username: str, password: str):
|
|
username = username.strip().lower()
|
|
if len(username) < 3:
|
|
raise ValueError("username must have 3 characters")
|
|
environment = os.getenv("APP_ENV") or settings.environment
|
|
weak_passwords_allowed = os.getenv("ALLOW_WEAK_DEV_PASSWORD", "0") == "1"
|
|
if len(password) < 12 and (environment.strip().lower() == "production" or not weak_passwords_allowed):
|
|
raise ValueError("password must have 12 characters")
|
|
if session.scalar(select(AdminUser).where(AdminUser.username == username)):
|
|
raise ValueError("username already exists")
|
|
admin = AdminUser(username=username, password_hash=hash_password(password))
|
|
session.add(admin); session.commit(); return admin
|
|
|
|
|
|
def create_session(session, admin: AdminUser, request: Request, seconds: int | None = None):
|
|
raw_token = secrets.token_urlsafe(32)
|
|
raw_csrf = secrets.token_urlsafe(32)
|
|
now = datetime.now(timezone.utc)
|
|
lifetime = seconds if seconds is not None else settings.web_session_seconds
|
|
row = WebSession(admin_user_id=admin.id, token_hash=_digest(raw_token), csrf_hash=_digest(raw_csrf), expires_at=now + timedelta(seconds=lifetime),user_agent=request.headers.get("user-agent", "")[:512], remote_addr=request.client.host if request.client else None)
|
|
session.add(row); admin.last_login_at = now; session.commit()
|
|
return raw_token, raw_csrf
|
|
|
|
|
|
def current_session(request: Request, session):
|
|
token = request.cookies.get(settings.web_cookie_name)
|
|
if not token: return None, None
|
|
row = session.scalar(select(WebSession).where(WebSession.token_hash == _digest(token)))
|
|
if not row or row.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc): return None, None
|
|
admin = session.get(AdminUser, row.admin_user_id)
|
|
if not admin or not admin.is_active: return None, None
|
|
return row, admin
|
|
|
|
|
|
def require_human(request: Request, session):
|
|
web_session, admin = current_session(request, session)
|
|
if not admin: raise HTTPException(401, "human login required")
|
|
return web_session, admin
|
|
|
|
|
|
def require_csrf(web_session: WebSession, csrf_token: str | None):
|
|
if not csrf_token or not hmac.compare_digest(web_session.csrf_hash, _digest(csrf_token)):
|
|
raise HTTPException(403, "invalid CSRF token")
|