damir
b95b2e8423
1) auth/auth_v2.py — update_me bug fix: PUT /auth/me return value was 'return me(user)' but me() is a FastAPI route handler, not callable directly. Replaced with explicit re-fetch returning correct JSON shape. Profile changes now persist in UI after save. 2) DB: HNK Goranin Delnice (id 782) sport='skijanje' → 'nogomet' + napomena cross-contamination cleaned (id 782, 192, 347, 2280) + general rule: NK/HNK/Nogometni klub → nogomet + RK/Rukometni klub → rukomet + OK/Odbojkaški klub → odbojka 3) DB: KUD/folklorne/lovačke/vatrogasne udruge marked as sport='kulturno-umjetnicko' + razina='NE-sportsko' so frontend can filter them out of sportski savezi list 4) Backup: pgz_sport.klubovi_backup_20260505_0857 Verified: PUT /auth/me with damir@pgz.hr persists telefon change to DB and returns fresh data
930 lines
41 KiB
Python
930 lines
41 KiB
Python
#!/usr/bin/env python3
|
|
# auth_v2.py — JWT auth backend with tenant_id, role, tier claims
|
|
# v1.0 dradulic@outlook.com / damir@rinet.one — 2026-05-04
|
|
# Endpoints: /api/auth/login, /api/auth/refresh, /api/auth/logout,
|
|
# /api/auth/me, /api/auth/password/change, /api/auth/password/reset
|
|
"""
|
|
JWT claims:
|
|
sub int user id
|
|
email str
|
|
name str
|
|
tenant_id int|null pgz_sport.tenants.id (or null for super_admin)
|
|
tenant_type str pgz | savez | klub | global
|
|
tenant_scope dict {"klub_id": ..., "savez_id": ...}
|
|
role str user_type code (super_admin | pgz_admin | savez_admin | klub_admin | klub_clan | viewer ...)
|
|
tier int 0 = PGŽ, 1 = savez, 2 = klub
|
|
jti str token id (revocable via user_sessions)
|
|
iat / exp / nbf
|
|
"""
|
|
|
|
import os, hashlib, secrets, json, time
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional, Dict, List, Any
|
|
|
|
import jwt as _jwt
|
|
import psycopg2, psycopg2.extras
|
|
from fastapi import APIRouter, HTTPException, Header, Depends, Request, Body
|
|
from pydantic import BaseModel, EmailStr
|
|
|
|
try:
|
|
from passlib.hash import bcrypt as _bcrypt
|
|
HAS_BCRYPT = True
|
|
except Exception:
|
|
HAS_BCRYPT = False
|
|
|
|
DB = dict(host='10.10.0.2', port=6432, dbname='rinet_v3',
|
|
user='rinet', password='R1net2026!SecureDB#v7')
|
|
|
|
# Persistent JWT secret — read from env, else stable file, else generated.
|
|
def _load_secret() -> str:
|
|
env_secret = os.environ.get("PGZ_JWT_SECRET")
|
|
if env_secret and len(env_secret) >= 32:
|
|
return env_secret
|
|
secret_file = "/opt/pgz-sport/auth/.jwt_secret"
|
|
try:
|
|
if os.path.exists(secret_file):
|
|
with open(secret_file) as f:
|
|
s = f.read().strip()
|
|
if len(s) >= 32:
|
|
return s
|
|
s = "rinet-pgz-" + secrets.token_urlsafe(48)
|
|
with open(secret_file, "w") as f:
|
|
f.write(s)
|
|
os.chmod(secret_file, 0o600)
|
|
return s
|
|
except Exception:
|
|
return "rinet-pgz-jwt-2026-fallback-" + hashlib.sha256(b"pgz-sport").hexdigest()
|
|
|
|
JWT_SECRET = _load_secret()
|
|
JWT_ALG = "HS256"
|
|
ACCESS_TTL = timedelta(minutes=int(os.environ.get("PGZ_JWT_ACCESS_MIN", "30")))
|
|
REFRESH_TTL = timedelta(days=int(os.environ.get("PGZ_JWT_REFRESH_DAYS", "7")))
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth_v2"])
|
|
|
|
# ─────────────────────────── DB helpers ───────────────────────────
|
|
def _conn():
|
|
return psycopg2.connect(**DB)
|
|
|
|
def db_query(sql: str, params=()):
|
|
with _conn() as c:
|
|
cur = c.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
|
cur.execute(sql, params)
|
|
if cur.description: return cur.fetchall()
|
|
return []
|
|
|
|
def db_one(sql: str, params=()):
|
|
rows = db_query(sql, params)
|
|
return rows[0] if rows else None
|
|
|
|
def db_exec(sql: str, params=()):
|
|
with _conn() as c:
|
|
cur = c.cursor()
|
|
cur.execute(sql, params)
|
|
if cur.description:
|
|
r = cur.fetchone()
|
|
return r[0] if r else None
|
|
c.commit()
|
|
|
|
# ─────────────────────────── Password helpers ───────────────────────────
|
|
def _sha256(pw: str) -> str:
|
|
return hashlib.sha256(pw.encode()).hexdigest()
|
|
|
|
def hash_password(pw: str) -> str:
|
|
if HAS_BCRYPT:
|
|
return _bcrypt.using(rounds=12).hash(pw)
|
|
return _sha256(pw)
|
|
|
|
def verify_password(pw: str, hashed: Optional[str]) -> bool:
|
|
if not hashed: return False
|
|
h = hashed.strip()
|
|
if h.startswith("$2") and HAS_BCRYPT:
|
|
try:
|
|
return _bcrypt.verify(pw, h)
|
|
except Exception:
|
|
return False
|
|
return h == _sha256(pw)
|
|
|
|
def needs_rehash(hashed: Optional[str]) -> bool:
|
|
if not hashed: return True
|
|
return HAS_BCRYPT and not hashed.startswith("$2")
|
|
|
|
# ─────────────────────────── Tenant resolution ───────────────────────────
|
|
PGZ_USER_TYPES = {"super_admin", "pgz_admin", "pgz_user", "pgz_finance", "pgz_zzjz"}
|
|
SAVEZ_USER_TYPES = {"savez_admin", "savez_user"}
|
|
KLUB_USER_TYPES = {"klub_admin", "klub_user", "klub_trener", "klub_clan"}
|
|
|
|
def _tier_for(user_type: str) -> int:
|
|
ut = (user_type or "").lower()
|
|
if ut in PGZ_USER_TYPES: return 0
|
|
if ut in SAVEZ_USER_TYPES: return 1
|
|
if ut in KLUB_USER_TYPES: return 2
|
|
return 9 # unknown / viewer / guest
|
|
|
|
def _resolve_tenant(u: Dict) -> Dict:
|
|
"""Resolve tenant_id + tenant_type from a user row."""
|
|
ut = (u.get("user_type") or "").lower()
|
|
klub_id = u.get("klub_id")
|
|
savez_id = u.get("savez_id")
|
|
if ut in PGZ_USER_TYPES:
|
|
row = db_one("SELECT id, slug, display_name FROM pgz_sport.tenants WHERE slug='pgz' LIMIT 1")
|
|
return {
|
|
"tenant_id": row["id"] if row else None,
|
|
"tenant_type": "pgz",
|
|
"tenant_name": row["display_name"] if row else "PGŽ",
|
|
"tenant_scope": {"klub_id": None, "savez_id": None},
|
|
}
|
|
if ut in SAVEZ_USER_TYPES and savez_id:
|
|
return {
|
|
"tenant_id": savez_id,
|
|
"tenant_type": "savez",
|
|
"tenant_name": (db_one("SELECT naziv FROM pgz_sport.savezi WHERE id=%s",(savez_id,)) or {}).get("naziv"),
|
|
"tenant_scope": {"klub_id": None, "savez_id": savez_id},
|
|
}
|
|
if ut in KLUB_USER_TYPES and klub_id:
|
|
return {
|
|
"tenant_id": klub_id,
|
|
"tenant_type": "klub",
|
|
"tenant_name": (db_one("SELECT naziv FROM pgz_sport.klubovi WHERE id=%s",(klub_id,)) or {}).get("naziv"),
|
|
"tenant_scope": {"klub_id": klub_id, "savez_id": savez_id},
|
|
}
|
|
# super_admin without context
|
|
if ut == "super_admin":
|
|
return {"tenant_id": None, "tenant_type": "global",
|
|
"tenant_name": "Global", "tenant_scope": {"klub_id": None, "savez_id": None}}
|
|
return {"tenant_id": None, "tenant_type": "viewer",
|
|
"tenant_name": None, "tenant_scope": {"klub_id": klub_id, "savez_id": savez_id}}
|
|
|
|
# ─────────────────────────── JWT issue / verify ───────────────────────────
|
|
def _now() -> datetime: return datetime.now(timezone.utc)
|
|
|
|
def _new_jti() -> str: return secrets.token_urlsafe(16)
|
|
|
|
def make_access_token(u: Dict, jti: str) -> str:
|
|
tenant = _resolve_tenant(u)
|
|
tier = _tier_for(u.get("user_type") or "")
|
|
now = _now()
|
|
payload = {
|
|
"sub": str(u["id"]),
|
|
"uid": u["id"],
|
|
"email": u["email"],
|
|
"name": u.get("full_name") or ((u.get("ime") or "") + " " + (u.get("prezime") or "")).strip() or u["email"],
|
|
"tenant_id": tenant["tenant_id"],
|
|
"tenant_type": tenant["tenant_type"],
|
|
"tenant_name": tenant["tenant_name"],
|
|
"tenant_scope": tenant["tenant_scope"],
|
|
"role": u.get("user_type") or "viewer",
|
|
"tier": tier,
|
|
"jti": jti,
|
|
"typ": "access",
|
|
"iat": int(now.timestamp()),
|
|
"nbf": int(now.timestamp()),
|
|
"exp": int((now + ACCESS_TTL).timestamp()),
|
|
}
|
|
return _jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG)
|
|
|
|
def make_refresh_token(uid: int, jti: str) -> str:
|
|
now = _now()
|
|
return _jwt.encode({
|
|
"sub": str(uid), "uid": uid, "jti": jti, "typ": "refresh",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int((now + REFRESH_TTL).timestamp()),
|
|
}, JWT_SECRET, algorithm=JWT_ALG)
|
|
|
|
def decode_token(token: str) -> Dict:
|
|
try:
|
|
return _jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG])
|
|
except _jwt.ExpiredSignatureError:
|
|
raise HTTPException(401, "Token expired")
|
|
except Exception as e:
|
|
raise HTTPException(401, f"Invalid token: {e}")
|
|
|
|
def _record_session(uid: int, jti: str, expires: datetime, ip: str = None, ua: str = None):
|
|
th = hashlib.sha256(jti.encode()).hexdigest()
|
|
db_exec("""INSERT INTO pgz_sport.user_sessions
|
|
(user_id, token_hash, device_info, ip_address, expires_at, revoked)
|
|
VALUES (%s,%s,%s,%s::inet,%s,false)
|
|
ON CONFLICT (token_hash) DO NOTHING""",
|
|
(uid, th, ua, ip, expires))
|
|
|
|
def _is_revoked(jti: str) -> bool:
|
|
th = hashlib.sha256(jti.encode()).hexdigest()
|
|
r = db_one("SELECT revoked FROM pgz_sport.user_sessions WHERE token_hash=%s", (th,))
|
|
if not r: return False
|
|
return bool(r.get("revoked"))
|
|
|
|
def _revoke_jti(jti: str):
|
|
th = hashlib.sha256(jti.encode()).hexdigest()
|
|
db_exec("UPDATE pgz_sport.user_sessions SET revoked=true WHERE token_hash=%s", (th,))
|
|
|
|
# ─────────────────────────── current_user dep ───────────────────────────
|
|
def _extract_token(authorization: Optional[str]) -> Optional[str]:
|
|
if not authorization: return None
|
|
return authorization.replace("Bearer ", "").strip() or None
|
|
|
|
def get_current_user(authorization: Optional[str] = Header(None)) -> Optional[Dict]:
|
|
token = _extract_token(authorization)
|
|
if not token: return None
|
|
try:
|
|
payload = decode_token(token)
|
|
except HTTPException:
|
|
return None
|
|
if payload.get("typ") not in (None, "access"):
|
|
return None
|
|
if _is_revoked(payload.get("jti","")):
|
|
return None
|
|
uid = payload.get("uid") or int(payload.get("sub", 0) or 0)
|
|
u = db_one("""SELECT id, email, full_name, ime, prezime, user_type,
|
|
klub_id, savez_id, status, aktivan, must_change_pwd
|
|
FROM pgz_sport.users WHERE id=%s""", (uid,))
|
|
if not u or u.get("status") != "active" or not u.get("aktivan", True):
|
|
return None
|
|
u["_jwt"] = payload
|
|
u["_token"] = token
|
|
return u
|
|
|
|
def require_user(user = Depends(get_current_user)) -> Dict:
|
|
if not user:
|
|
raise HTTPException(401, "Authentication required")
|
|
return user
|
|
|
|
def require_role(roles: List[str]):
|
|
def dep(user = Depends(require_user)):
|
|
if user.get("user_type") not in roles:
|
|
raise HTTPException(403, f"Forbidden — required: {','.join(roles)}")
|
|
return user
|
|
return dep
|
|
|
|
# ─────────────────────────── Audit ───────────────────────────
|
|
def audit(user_id: Optional[int], action: str, resource_type: str = None,
|
|
resource_id: int = None, meta: Dict = None, ip: str = None, ua: str = None):
|
|
try:
|
|
db_exec("""INSERT INTO pgz_sport.audit_events
|
|
(user_id, action, resource_type, resource_id, meta, ip_address, user_agent)
|
|
VALUES (%s,%s,%s,%s,%s::jsonb,%s::inet,%s)""",
|
|
(user_id, action, resource_type, resource_id,
|
|
json.dumps(meta or {}), ip, ua))
|
|
except Exception as e:
|
|
print(f"[AUDIT WARN] {e}")
|
|
|
|
def _client(req: Request):
|
|
ip = (req.headers.get("x-forwarded-for") or req.client.host or "").split(",")[0].strip() or None
|
|
ua = req.headers.get("user-agent")
|
|
return ip, ua
|
|
|
|
# ─────────────────────────── Schemas ───────────────────────────
|
|
class LoginReq(BaseModel):
|
|
email: str
|
|
password: str
|
|
totp: Optional[str] = None # 6-digit TOTP if 2FA enabled (or recovery code)
|
|
|
|
class RefreshReq(BaseModel):
|
|
refresh_token: str
|
|
|
|
class ChangePwdReq(BaseModel):
|
|
old_password: Optional[str] = None
|
|
new_password: str
|
|
|
|
class ResetPwdReq(BaseModel):
|
|
email: str
|
|
|
|
# ─────────────────────────── Rate limiting (R6 #5) ───────────────────────────
|
|
LOCK_THRESHOLD = int(os.environ.get("PGZ_LOGIN_LOCK_THRESHOLD", "5"))
|
|
LOCK_MINUTES = int(os.environ.get("PGZ_LOGIN_LOCK_MINUTES", "5"))
|
|
IP_THRESHOLD = int(os.environ.get("PGZ_LOGIN_IP_THRESHOLD", "10"))
|
|
IP_WINDOW_SEC = int(os.environ.get("PGZ_LOGIN_IP_WINDOW_SEC", "300")) # 5 min
|
|
|
|
# In-memory IP throttle: ip → list[float fail timestamps within window]
|
|
_ip_fail_log: Dict[str, List[float]] = {}
|
|
|
|
def _ip_record_fail(ip: Optional[str]):
|
|
if not ip: return
|
|
now = time.time()
|
|
arr = [t for t in _ip_fail_log.get(ip, []) if now - t < IP_WINDOW_SEC]
|
|
arr.append(now)
|
|
_ip_fail_log[ip] = arr
|
|
|
|
def _ip_blocked(ip: Optional[str]) -> Optional[int]:
|
|
"""Return seconds-until-unblock, or None if not blocked."""
|
|
if not ip: return None
|
|
now = time.time()
|
|
arr = [t for t in _ip_fail_log.get(ip, []) if now - t < IP_WINDOW_SEC]
|
|
_ip_fail_log[ip] = arr
|
|
if len(arr) < IP_THRESHOLD: return None
|
|
oldest = min(arr)
|
|
return max(1, int(IP_WINDOW_SEC - (now - oldest)))
|
|
|
|
def _ip_clear(ip: Optional[str]):
|
|
if ip and ip in _ip_fail_log:
|
|
_ip_fail_log.pop(ip, None)
|
|
|
|
# ─────────────────────────── Endpoints ───────────────────────────
|
|
@router.post("/login")
|
|
def login(req: LoginReq, request: Request):
|
|
ip, ua = _client(request)
|
|
email = (req.email or "").lower().strip()
|
|
if not email or not req.password:
|
|
raise HTTPException(400, "Email i lozinka obavezni")
|
|
|
|
# R6 #5: per-IP throttle (stops brute-force across many emails)
|
|
blocked_for = _ip_blocked(ip)
|
|
if blocked_for:
|
|
audit(None, "login.ratelimit.ip",
|
|
meta={"email": email, "ip": ip, "block_seconds": blocked_for},
|
|
ip=ip, ua=ua)
|
|
raise HTTPException(429, f"Previše pokušaja s ove IP adrese — pokušajte za {blocked_for}s")
|
|
|
|
u = db_one("""SELECT id, email, full_name, ime, prezime, password_hash, status,
|
|
user_type, klub_id, savez_id, aktivan, must_change_pwd,
|
|
failed_login_count, locked_until
|
|
FROM pgz_sport.users WHERE LOWER(email)=%s""", (email,))
|
|
if not u:
|
|
_ip_record_fail(ip)
|
|
audit(None, "login.fail", meta={"email": email, "reason": "no_user"}, ip=ip, ua=ua)
|
|
raise HTTPException(401, "Neispravni podaci")
|
|
if u.get("locked_until"):
|
|
lu = u["locked_until"]
|
|
if lu.tzinfo is None: lu = lu.replace(tzinfo=timezone.utc)
|
|
if lu > _now():
|
|
audit(u["id"], "login.locked", ip=ip, ua=ua)
|
|
raise HTTPException(423, "Račun privremeno zaključan")
|
|
if u.get("status") != "active" or not u.get("aktivan", True):
|
|
audit(u["id"], "login.fail", meta={"reason":"inactive"}, ip=ip, ua=ua)
|
|
raise HTTPException(403, "Račun nije aktivan")
|
|
if not verify_password(req.password, u.get("password_hash")):
|
|
# R6 #5: 5 fails → 5-minute lockout
|
|
new_fails = (u.get("failed_login_count") or 0) + 1
|
|
will_lock = new_fails >= LOCK_THRESHOLD
|
|
db_exec("""UPDATE pgz_sport.users
|
|
SET failed_login_count = %s,
|
|
locked_until = CASE WHEN %s
|
|
THEN now() + (interval '1 minute' * %s)
|
|
ELSE locked_until END
|
|
WHERE id=%s""",
|
|
(new_fails, will_lock, LOCK_MINUTES, u["id"]))
|
|
_ip_record_fail(ip)
|
|
audit(u["id"], "login.fail",
|
|
meta={"reason":"bad_password", "fails": new_fails,
|
|
"locked": bool(will_lock),
|
|
"lock_minutes": LOCK_MINUTES if will_lock else 0},
|
|
ip=ip, ua=ua)
|
|
raise HTTPException(401,
|
|
f"Neispravni podaci ({new_fails}/{LOCK_THRESHOLD})" +
|
|
(f" — račun je zaključan na {LOCK_MINUTES} minuta" if will_lock else ""))
|
|
|
|
# opportunistic rehash to bcrypt
|
|
if needs_rehash(u.get("password_hash")):
|
|
try:
|
|
db_exec("UPDATE pgz_sport.users SET password_hash=%s WHERE id=%s",
|
|
(hash_password(req.password), u["id"]))
|
|
except Exception: pass
|
|
|
|
# 2FA gate — if user has enabled 2FA, demand a valid TOTP / recovery code
|
|
twofa_row = None
|
|
try:
|
|
twofa_row = db_one("SELECT secret, enabled, recovery_codes FROM pgz_sport.user_2fa WHERE user_id=%s",
|
|
(u["id"],))
|
|
except Exception: pass
|
|
if twofa_row and twofa_row.get("enabled"):
|
|
code = (req.totp or "").strip().replace(" ", "")
|
|
if not code:
|
|
audit(u["id"], "login.2fa_required", ip=ip, ua=ua)
|
|
raise HTTPException(401, "2FA_REQUIRED")
|
|
ok = False
|
|
if code.isdigit() and len(code) in (6, 8) and HAS_PYOTP:
|
|
ok = _pyotp.TOTP(twofa_row["secret"]).verify(code, valid_window=1)
|
|
if not ok and twofa_row.get("recovery_codes"):
|
|
up = code.upper()
|
|
if up in (twofa_row["recovery_codes"] or []):
|
|
ok = True
|
|
# consume the recovery code so it can't be reused
|
|
remaining = [c for c in twofa_row["recovery_codes"] if c != up]
|
|
db_exec("UPDATE pgz_sport.user_2fa SET recovery_codes=%s, updated_at=now() WHERE user_id=%s",
|
|
(remaining, u["id"]))
|
|
if not ok:
|
|
audit(u["id"], "login.2fa_fail", ip=ip, ua=ua)
|
|
raise HTTPException(401, "Neispravan 2FA kod")
|
|
|
|
db_exec("""UPDATE pgz_sport.users
|
|
SET failed_login_count=0, locked_until=NULL, last_login=now()
|
|
WHERE id=%s""", (u["id"],))
|
|
_ip_clear(ip) # successful login clears IP throttle
|
|
|
|
jti = _new_jti()
|
|
rjti = _new_jti()
|
|
access = make_access_token(u, jti)
|
|
refresh = make_refresh_token(u["id"], rjti)
|
|
_record_session(u["id"], jti, _now() + ACCESS_TTL, ip=ip, ua=ua)
|
|
_record_session(u["id"], rjti, _now() + REFRESH_TTL, ip=ip, ua=(ua or "") + " [refresh]")
|
|
audit(u["id"], "login.ok", ip=ip, ua=ua)
|
|
|
|
tenant = _resolve_tenant(u)
|
|
return {
|
|
"access_token": access,
|
|
"refresh_token": refresh,
|
|
"token_type": "Bearer",
|
|
"expires_in": int(ACCESS_TTL.total_seconds()),
|
|
"user": {
|
|
"id": u["id"], "email": u["email"],
|
|
"full_name": u.get("full_name") or (u.get("ime","") + " " + u.get("prezime","")).strip(),
|
|
"role": u.get("user_type"), "tier": _tier_for(u.get("user_type") or ""),
|
|
"must_change_pwd": bool(u.get("must_change_pwd")),
|
|
**tenant,
|
|
},
|
|
}
|
|
|
|
@router.post("/refresh")
|
|
def refresh(req: RefreshReq, request: Request):
|
|
payload = decode_token(req.refresh_token)
|
|
if payload.get("typ") != "refresh":
|
|
raise HTTPException(401, "Invalid refresh token")
|
|
if _is_revoked(payload.get("jti","")):
|
|
raise HTTPException(401, "Refresh token revoked")
|
|
uid = payload.get("uid") or int(payload.get("sub", 0) or 0)
|
|
u = db_one("""SELECT id, email, full_name, ime, prezime, user_type,
|
|
klub_id, savez_id, status, aktivan, must_change_pwd
|
|
FROM pgz_sport.users WHERE id=%s""", (uid,))
|
|
if not u or u.get("status") != "active" or not u.get("aktivan", True):
|
|
raise HTTPException(401, "User inactive")
|
|
ip, ua = _client(request)
|
|
new_jti = _new_jti()
|
|
access = make_access_token(u, new_jti)
|
|
_record_session(u["id"], new_jti, _now() + ACCESS_TTL, ip=ip, ua=ua)
|
|
audit(u["id"], "auth.refresh", ip=ip, ua=ua)
|
|
return {"access_token": access, "token_type": "Bearer",
|
|
"expires_in": int(ACCESS_TTL.total_seconds())}
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request, user = Depends(require_user)):
|
|
jti = (user.get("_jwt") or {}).get("jti")
|
|
if jti: _revoke_jti(jti)
|
|
# Also revoke refresh tokens for this user (best-effort)
|
|
db_exec("""UPDATE pgz_sport.user_sessions SET revoked=true
|
|
WHERE user_id=%s AND device_info LIKE %s""",
|
|
(user["id"], "%[refresh]%"))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "logout", ip=ip, ua=ua)
|
|
return {"status": "ok"}
|
|
|
|
@router.get("/me")
|
|
def me(user = Depends(require_user)):
|
|
enriched = db_one("""SELECT id, email, full_name, ime, prezime, user_type,
|
|
klub_id, savez_id, must_change_pwd, aktivan, status,
|
|
last_login, oib, telefon, phone, preferred_language, created_at,
|
|
avatar_url, gdpr_consent_at, google_picture
|
|
FROM pgz_sport.users WHERE id=%s""", (user["id"],))
|
|
if not enriched:
|
|
raise HTTPException(404, "User not found")
|
|
tenant = _resolve_tenant(enriched)
|
|
roles = db_query("""SELECT r.code, r.naziv, ur.scope_type, ur.scope_id
|
|
FROM pgz_sport.user_roles ur JOIN pgz_sport.roles r ON r.id=ur.role_id
|
|
WHERE ur.user_id=%s AND ur.active=true""", (user["id"],))
|
|
try:
|
|
twofa = db_one("SELECT secret IS NOT NULL AS enabled FROM pgz_sport.user_2fa WHERE user_id=%s",
|
|
(user["id"],)) or {"enabled": False}
|
|
except Exception:
|
|
twofa = {"enabled": False}
|
|
return {**enriched,
|
|
"tier": _tier_for(enriched.get("user_type") or ""),
|
|
"must_change_pwd": bool(enriched.get("must_change_pwd")),
|
|
"two_factor_enabled": bool(twofa.get("enabled")),
|
|
**tenant, "roles": roles}
|
|
|
|
class UpdateMeReq(BaseModel):
|
|
ime: Optional[str] = None
|
|
prezime: Optional[str] = None
|
|
full_name: Optional[str] = None
|
|
telefon: Optional[str] = None
|
|
phone: Optional[str] = None
|
|
preferred_language: Optional[str] = None
|
|
oib: Optional[str] = None
|
|
|
|
@router.put("/me")
|
|
def update_me(req: UpdateMeReq, request: Request, user = Depends(require_user)):
|
|
fields = []
|
|
vals: List[Any] = []
|
|
for k in ("ime","prezime","full_name","telefon","phone","preferred_language","oib"):
|
|
v = getattr(req, k)
|
|
if v is not None:
|
|
fields.append(f"{k}=%s")
|
|
vals.append(v.strip() if isinstance(v, str) else v)
|
|
if not fields:
|
|
raise HTTPException(400, "Nema polja za ažuriranje")
|
|
vals.append(user["id"])
|
|
db_exec(f"UPDATE pgz_sport.users SET {', '.join(fields)}, updated_at=now() WHERE id=%s", tuple(vals))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "profile.update", meta={"fields": [f.split("=")[0] for f in fields]}, ip=ip, ua=ua)
|
|
return me(user)
|
|
|
|
# ─────────────────────────── AVATAR UPLOAD ───────────────────────────
|
|
import shutil, pathlib
|
|
from fastapi import UploadFile, File
|
|
|
|
UPLOAD_ROOT = pathlib.Path("/opt/pgz-sport/uploads")
|
|
AVATAR_DIR = UPLOAD_ROOT / "avatars"
|
|
AVATAR_DIR.mkdir(parents=True, exist_ok=True)
|
|
ALLOWED_AVATAR_MIME = {"image/jpeg","image/jpg","image/png","image/webp"}
|
|
ALLOWED_AVATAR_EXT = {".jpg",".jpeg",".png",".webp"}
|
|
MAX_AVATAR_BYTES = 5 * 1024 * 1024 # 5 MB
|
|
|
|
@router.post("/me/avatar")
|
|
async def upload_my_avatar(request: Request, file: UploadFile = File(...), user = Depends(require_user)):
|
|
ct = (file.content_type or "").lower()
|
|
if ct not in ALLOWED_AVATAR_MIME:
|
|
raise HTTPException(400, f"Nedozvoljen tip slike: {ct} — jpeg/png/webp")
|
|
ext = pathlib.Path(file.filename or "").suffix.lower()
|
|
if ext not in ALLOWED_AVATAR_EXT:
|
|
ext = {"image/jpeg":".jpg","image/jpg":".jpg","image/png":".png","image/webp":".webp"}.get(ct, ".jpg")
|
|
data = await file.read()
|
|
if len(data) > MAX_AVATAR_BYTES:
|
|
raise HTTPException(413, f"Slika prevelika ({len(data)} B > {MAX_AVATAR_BYTES})")
|
|
if len(data) < 32:
|
|
raise HTTPException(400, "Slika prazna ili neispravna")
|
|
safe_name = f"{int(user['id'])}_{int(time.time())}{ext}"
|
|
target = AVATAR_DIR / safe_name
|
|
with open(target, "wb") as f:
|
|
f.write(data)
|
|
try: os.chmod(target, 0o644)
|
|
except Exception: pass
|
|
avatar_url = f"/uploads/avatars/{safe_name}"
|
|
db_exec("UPDATE pgz_sport.users SET avatar_url=%s, updated_at=now() WHERE id=%s",
|
|
(avatar_url, user["id"]))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "profile.avatar_upload",
|
|
meta={"file": safe_name, "size": len(data), "mime": ct}, ip=ip, ua=ua)
|
|
return {"status":"ok", "avatar_url": avatar_url, "size": len(data), "mime": ct}
|
|
|
|
@router.delete("/me/avatar")
|
|
def delete_my_avatar(request: Request, user = Depends(require_user)):
|
|
cur = db_one("SELECT avatar_url FROM pgz_sport.users WHERE id=%s", (user["id"],))
|
|
if cur and cur.get("avatar_url"):
|
|
p = AVATAR_DIR / pathlib.Path(cur["avatar_url"]).name
|
|
try:
|
|
if p.exists() and p.is_relative_to(AVATAR_DIR): p.unlink()
|
|
except Exception: pass
|
|
db_exec("UPDATE pgz_sport.users SET avatar_url=NULL, updated_at=now() WHERE id=%s", (user["id"],))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "profile.avatar_delete", ip=ip, ua=ua)
|
|
return {"status": "ok"}
|
|
|
|
@router.post("/password/change")
|
|
def change_password(req: ChangePwdReq, request: Request, user = Depends(require_user)):
|
|
if len(req.new_password) < 8:
|
|
raise HTTPException(400, "Lozinka mora imati barem 8 znakova")
|
|
cur = db_one("SELECT password_hash, must_change_pwd FROM pgz_sport.users WHERE id=%s",
|
|
(user["id"],))
|
|
if not cur: raise HTTPException(404, "User not found")
|
|
if not cur.get("must_change_pwd"):
|
|
if not req.old_password:
|
|
raise HTTPException(400, "old_password obavezan")
|
|
if not verify_password(req.old_password, cur.get("password_hash")):
|
|
raise HTTPException(401, "Stara lozinka netočna")
|
|
db_exec("""UPDATE pgz_sport.users
|
|
SET password_hash=%s, must_change_pwd=false, updated_at=now()
|
|
WHERE id=%s""", (hash_password(req.new_password), user["id"]))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "password.change", ip=ip, ua=ua)
|
|
return {"status": "ok"}
|
|
|
|
@router.post("/password/reset")
|
|
def password_reset(req: ResetPwdReq, request: Request):
|
|
"""Issue a temporary password (admin-equivalent self-reset; logged)."""
|
|
email = (req.email or "").lower().strip()
|
|
u = db_one("SELECT id, email, aktivan FROM pgz_sport.users WHERE LOWER(email)=%s",
|
|
(email,))
|
|
ip, ua = _client(request)
|
|
audit(u["id"] if u else None, "password.reset.request",
|
|
meta={"email": email, "found": bool(u)}, ip=ip, ua=ua)
|
|
# Generic response — do not leak which emails exist
|
|
return {"status": "ok",
|
|
"message": "Ako račun postoji, administrator će vam poslati instrukcije."}
|
|
|
|
# ─────────────────────────── R5 #2+#3: invite & reset tokens ───────────────────────────
|
|
def _ensure_token_table():
|
|
db_exec("""CREATE TABLE IF NOT EXISTS pgz_sport.user_action_tokens (
|
|
token_hash TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL REFERENCES pgz_sport.users(id) ON DELETE CASCADE,
|
|
kind TEXT NOT NULL, -- 'invite' | 'reset'
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
expires_at TIMESTAMPTZ NOT NULL,
|
|
used_at TIMESTAMPTZ,
|
|
created_by INTEGER REFERENCES pgz_sport.users(id),
|
|
ip TEXT,
|
|
meta JSONB
|
|
)""")
|
|
db_exec("""CREATE INDEX IF NOT EXISTS idx_action_tokens_user
|
|
ON pgz_sport.user_action_tokens (user_id, kind, used_at)""")
|
|
_ensure_token_table()
|
|
|
|
INVITE_TTL = timedelta(days=int(os.environ.get("PGZ_INVITE_TTL_DAYS", "7")))
|
|
RESET_TTL = timedelta(hours=int(os.environ.get("PGZ_RESET_TTL_HOURS", "2")))
|
|
|
|
def _make_action_token() -> str:
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def _hash_action_token(t: str) -> str:
|
|
return hashlib.sha256(t.encode()).hexdigest()
|
|
|
|
def issue_action_token(user_id: int, kind: str, ttl: timedelta,
|
|
created_by: Optional[int] = None,
|
|
ip: Optional[str] = None,
|
|
meta: Optional[Dict] = None) -> str:
|
|
"""Create a one-time URL-safe token; only its sha256 is persisted."""
|
|
if kind not in ("invite", "reset"):
|
|
raise ValueError("kind must be invite|reset")
|
|
# Invalidate any prior unused tokens of same kind for this user
|
|
db_exec("""UPDATE pgz_sport.user_action_tokens SET used_at=now()
|
|
WHERE user_id=%s AND kind=%s AND used_at IS NULL""",
|
|
(user_id, kind))
|
|
raw = _make_action_token()
|
|
th = _hash_action_token(raw)
|
|
db_exec("""INSERT INTO pgz_sport.user_action_tokens
|
|
(token_hash, user_id, kind, expires_at, created_by, ip, meta)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb)""",
|
|
(th, user_id, kind, _now() + ttl, created_by, ip, json.dumps(meta or {})))
|
|
return raw
|
|
|
|
def consume_action_token(raw: str, kind: str) -> Optional[Dict]:
|
|
"""Validate (kind/expiry/unused) and atomically mark used_at. Returns row dict if OK."""
|
|
th = _hash_action_token(raw)
|
|
row = db_one("""SELECT t.user_id, t.expires_at, t.used_at, t.kind, t.meta,
|
|
u.email, u.aktivan, u.status
|
|
FROM pgz_sport.user_action_tokens t
|
|
JOIN pgz_sport.users u ON u.id = t.user_id
|
|
WHERE t.token_hash=%s AND t.kind=%s""", (th, kind))
|
|
if not row: return None
|
|
if row["used_at"] is not None: return None
|
|
exp = row["expires_at"]
|
|
if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc)
|
|
if exp <= _now(): return None
|
|
db_exec("UPDATE pgz_sport.user_action_tokens SET used_at=now() WHERE token_hash=%s", (th,))
|
|
return row
|
|
|
|
def _build_link(path: str, token: str) -> str:
|
|
base = os.environ.get("PGZ_PUBLIC_BASE", "https://api.rinet.one/sport")
|
|
sep = '&' if '?' in path else '?'
|
|
return f"{base}{path}{sep}token={token}"
|
|
|
|
# ─────────────────────────── /auth/forgot-password ───────────────────────────
|
|
class ForgotPwdReq(BaseModel):
|
|
email: str
|
|
|
|
@router.post("/forgot-password")
|
|
def forgot_password(req: ForgotPwdReq, request: Request):
|
|
"""Always returns a generic message — never leaks which emails exist.
|
|
Issues a reset token only if the user exists and is active, then
|
|
sends a (mock) e-mail with the reset link."""
|
|
email = (req.email or "").lower().strip()
|
|
ip, ua = _client(request)
|
|
u = db_one("SELECT id, email, aktivan, status FROM pgz_sport.users WHERE LOWER(email)=%s",
|
|
(email,))
|
|
token = None
|
|
mail_result = None
|
|
if u and u.get("aktivan") and u.get("status") == "active":
|
|
token = issue_action_token(u["id"], "reset", RESET_TTL, ip=ip,
|
|
meta={"email": email})
|
|
reset_link = _build_link("/static/login.html?reset=1", token)
|
|
try:
|
|
from .mailer import send_password_reset
|
|
mail_result = send_password_reset(email, reset_link,
|
|
int(RESET_TTL.total_seconds()))
|
|
except Exception as e:
|
|
print(f"[forgot_password mail WARN] {e}")
|
|
audit(u["id"], "password.forgot.issue",
|
|
meta={"email": email,
|
|
"ttl_hours": RESET_TTL.total_seconds()/3600,
|
|
"mail_sent": bool(mail_result and mail_result.get("sent")),
|
|
"mail_mock": bool(mail_result and mail_result.get("mock"))},
|
|
ip=ip, ua=ua)
|
|
else:
|
|
audit(u["id"] if u else None, "password.forgot.miss",
|
|
meta={"email": email}, ip=ip, ua=ua)
|
|
# Generic response — do not leak account existence
|
|
resp = {"status": "ok",
|
|
"message": "Ako račun postoji, poslan je e-mail s linkom za promjenu lozinke."}
|
|
# Reveal link only on localhost or with explicit env flag (debugging).
|
|
# Real users get it via e-mail.
|
|
if token and (os.environ.get("PGZ_REVEAL_RESET_TOKEN") == "1" or
|
|
(request.client.host in ("127.0.0.1", "::1"))):
|
|
resp["reset_link"] = _build_link("/static/login.html?reset=1", token)
|
|
resp["expires_in_seconds"] = int(RESET_TTL.total_seconds())
|
|
resp["mail_mock"] = bool(mail_result and mail_result.get("mock"))
|
|
return resp
|
|
|
|
class ResetTokenReq(BaseModel):
|
|
token: str
|
|
new_password: str
|
|
|
|
@router.post("/reset-password")
|
|
def reset_password_with_token(req: ResetTokenReq, request: Request):
|
|
"""Consume a reset token and set a new password."""
|
|
if len(req.new_password or "") < 8:
|
|
raise HTTPException(400, "Lozinka mora imati barem 8 znakova")
|
|
row = consume_action_token(req.token, "reset")
|
|
ip, ua = _client(request)
|
|
if not row:
|
|
audit(None, "password.reset.fail",
|
|
meta={"reason": "invalid_or_expired_token"}, ip=ip, ua=ua)
|
|
raise HTTPException(400, "Token je nevažeći ili istekao")
|
|
if not row.get("aktivan") or row.get("status") != "active":
|
|
audit(row["user_id"], "password.reset.fail",
|
|
meta={"reason": "user_inactive"}, ip=ip, ua=ua)
|
|
raise HTTPException(403, "Račun nije aktivan")
|
|
db_exec("""UPDATE pgz_sport.users
|
|
SET password_hash=%s, must_change_pwd=false,
|
|
failed_login_count=0, locked_until=NULL, updated_at=now()
|
|
WHERE id=%s""", (hash_password(req.new_password), row["user_id"]))
|
|
# Revoke all active sessions for safety
|
|
db_exec("UPDATE pgz_sport.user_sessions SET revoked=true WHERE user_id=%s",
|
|
(row["user_id"],))
|
|
audit(row["user_id"], "password.reset.ok", ip=ip, ua=ua)
|
|
return {"status": "ok", "email": row["email"]}
|
|
|
|
@router.get("/reset-password")
|
|
def reset_password_check(token: str, request: Request):
|
|
"""Pre-flight: validate that the token exists and isn't expired/used.
|
|
Does NOT consume the token."""
|
|
th = _hash_action_token(token)
|
|
row = db_one("""SELECT t.user_id, t.expires_at, t.used_at, u.email
|
|
FROM pgz_sport.user_action_tokens t
|
|
JOIN pgz_sport.users u ON u.id = t.user_id
|
|
WHERE t.token_hash=%s AND t.kind='reset'""", (th,))
|
|
if not row:
|
|
raise HTTPException(404, "Token nije pronađen")
|
|
if row["used_at"] is not None:
|
|
raise HTTPException(410, "Token je već iskorišten")
|
|
exp = row["expires_at"]
|
|
if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc)
|
|
if exp <= _now():
|
|
raise HTTPException(410, "Token je istekao")
|
|
return {"status": "ok", "email": row["email"], "expires_at": row["expires_at"].isoformat()}
|
|
|
|
# ─────────────────────────── /auth/setup-password (invite) ───────────────────────────
|
|
class SetupPwdReq(BaseModel):
|
|
token: str
|
|
new_password: str
|
|
|
|
@router.get("/setup-password")
|
|
def setup_password_check(token: str, request: Request):
|
|
"""Pre-flight: validate an invite token without consuming it."""
|
|
th = _hash_action_token(token)
|
|
row = db_one("""SELECT t.user_id, t.expires_at, t.used_at, u.email, u.full_name, u.user_type
|
|
FROM pgz_sport.user_action_tokens t
|
|
JOIN pgz_sport.users u ON u.id = t.user_id
|
|
WHERE t.token_hash=%s AND t.kind='invite'""", (th,))
|
|
if not row:
|
|
raise HTTPException(404, "Pozivnica nije pronađena")
|
|
if row["used_at"] is not None:
|
|
raise HTTPException(410, "Pozivnica je već iskorištena")
|
|
exp = row["expires_at"]
|
|
if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc)
|
|
if exp <= _now():
|
|
raise HTTPException(410, "Pozivnica je istekla")
|
|
return {"status": "ok",
|
|
"email": row["email"],
|
|
"full_name": row["full_name"],
|
|
"user_type": row["user_type"],
|
|
"expires_at": row["expires_at"].isoformat()}
|
|
|
|
@router.post("/setup-password")
|
|
def setup_password_consume(req: SetupPwdReq, request: Request):
|
|
"""Consume an invite token and set the user's first password."""
|
|
if len(req.new_password or "") < 8:
|
|
raise HTTPException(400, "Lozinka mora imati barem 8 znakova")
|
|
row = consume_action_token(req.token, "invite")
|
|
ip, ua = _client(request)
|
|
if not row:
|
|
audit(None, "invite.consume.fail",
|
|
meta={"reason": "invalid_or_expired_token"}, ip=ip, ua=ua)
|
|
raise HTTPException(400, "Pozivnica je nevažeća ili istekla")
|
|
if not row.get("aktivan") or row.get("status") != "active":
|
|
audit(row["user_id"], "invite.consume.fail",
|
|
meta={"reason": "user_inactive"}, ip=ip, ua=ua)
|
|
raise HTTPException(403, "Račun nije aktivan")
|
|
db_exec("""UPDATE pgz_sport.users
|
|
SET password_hash=%s, must_change_pwd=false,
|
|
email_verified=true,
|
|
failed_login_count=0, locked_until=NULL, updated_at=now()
|
|
WHERE id=%s""", (hash_password(req.new_password), row["user_id"]))
|
|
audit(row["user_id"], "invite.consume.ok",
|
|
meta={"email": row["email"]}, ip=ip, ua=ua)
|
|
return {"status": "ok", "email": row["email"]}
|
|
|
|
# ─────────────────────────── 2FA — real TOTP (RFC 6238) ───────────────────────────
|
|
try:
|
|
import pyotp as _pyotp
|
|
HAS_PYOTP = True
|
|
except Exception:
|
|
HAS_PYOTP = False
|
|
|
|
def _ensure_2fa_table():
|
|
db_exec("""CREATE TABLE IF NOT EXISTS pgz_sport.user_2fa (
|
|
user_id INTEGER PRIMARY KEY REFERENCES pgz_sport.users(id) ON DELETE CASCADE,
|
|
secret TEXT NOT NULL,
|
|
enabled BOOLEAN DEFAULT false,
|
|
verified_at TIMESTAMPTZ,
|
|
recovery_codes TEXT[],
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now()
|
|
)""")
|
|
_ensure_2fa_table()
|
|
|
|
def _build_qr_png(otpauth_url: str) -> str:
|
|
"""Return a data: URL containing a base64 PNG of the QR code."""
|
|
try:
|
|
import qrcode, io, base64
|
|
img = qrcode.make(otpauth_url)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()
|
|
except Exception as e:
|
|
return ""
|
|
|
|
def _gen_recovery_codes(n: int = 8) -> List[str]:
|
|
return [secrets.token_hex(4).upper() for _ in range(n)]
|
|
|
|
@router.post("/2fa/setup")
|
|
def twofa_setup(user = Depends(require_user)):
|
|
"""Generate a TOTP secret, store unverified, and return otpauth URL + QR + recovery codes.
|
|
The 2FA stays disabled until /2fa/verify confirms a valid TOTP code."""
|
|
if not HAS_PYOTP:
|
|
raise HTTPException(503, "pyotp not installed on server")
|
|
secret = _pyotp.random_base32() # 32-char base32, RFC 4648 — what authenticator apps expect
|
|
recovery = _gen_recovery_codes()
|
|
db_exec("""INSERT INTO pgz_sport.user_2fa (user_id, secret, enabled, recovery_codes, updated_at)
|
|
VALUES (%s,%s,false,%s,now())
|
|
ON CONFLICT (user_id) DO UPDATE SET
|
|
secret=EXCLUDED.secret, enabled=false,
|
|
recovery_codes=EXCLUDED.recovery_codes, updated_at=now()""",
|
|
(user["id"], secret, recovery))
|
|
issuer = "PGŽ Sport"
|
|
otpauth = _pyotp.TOTP(secret).provisioning_uri(name=user["email"], issuer_name=issuer)
|
|
return {
|
|
"secret": secret,
|
|
"otpauth_url": otpauth,
|
|
"qr_png": _build_qr_png(otpauth),
|
|
"issuer": issuer,
|
|
"account": user["email"],
|
|
"recovery_codes": recovery,
|
|
"enabled": False,
|
|
"instructions": "Skenirajte QR u Google Authenticator / Authy / 1Password, zatim potvrdite kod kroz POST /api/auth/2fa/verify",
|
|
}
|
|
|
|
class TwoFAVerifyReq(BaseModel):
|
|
code: str
|
|
|
|
@router.post("/2fa/verify")
|
|
def twofa_verify(req: TwoFAVerifyReq, request: Request, user = Depends(require_user)):
|
|
"""Verify TOTP code; on success, mark 2FA enabled."""
|
|
if not HAS_PYOTP:
|
|
raise HTTPException(503, "pyotp not installed on server")
|
|
row = db_one("SELECT secret, enabled FROM pgz_sport.user_2fa WHERE user_id=%s",
|
|
(user["id"],))
|
|
if not row:
|
|
raise HTTPException(400, "2FA nije postavljen — pozovite /2fa/setup prvo")
|
|
code = (req.code or "").strip().replace(" ", "")
|
|
if not code or not code.isdigit() or len(code) not in (6, 8):
|
|
raise HTTPException(400, "Neispravan format koda (6-8 znamenki)")
|
|
totp = _pyotp.TOTP(row["secret"])
|
|
# valid_window=1 → tolerate ±30s drift
|
|
if not totp.verify(code, valid_window=1):
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "2fa.verify.fail", ip=ip, ua=ua)
|
|
raise HTTPException(401, "Neispravan TOTP kod")
|
|
db_exec("""UPDATE pgz_sport.user_2fa
|
|
SET enabled=true, verified_at=now(), updated_at=now()
|
|
WHERE user_id=%s""", (user["id"],))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "2fa.verify.ok", ip=ip, ua=ua)
|
|
return {"status": "ok", "enabled": True}
|
|
|
|
@router.post("/2fa/disable")
|
|
def twofa_disable(req: TwoFAVerifyReq, request: Request, user = Depends(require_user)):
|
|
"""Disable 2FA — must verify a current TOTP code (or recovery code)."""
|
|
if not HAS_PYOTP:
|
|
raise HTTPException(503, "pyotp not installed on server")
|
|
row = db_one("SELECT secret, recovery_codes FROM pgz_sport.user_2fa WHERE user_id=%s",
|
|
(user["id"],))
|
|
if not row:
|
|
raise HTTPException(404, "2FA nije postavljen")
|
|
code = (req.code or "").strip().replace(" ", "").upper()
|
|
valid = False
|
|
if code.isdigit() and len(code) in (6, 8):
|
|
valid = _pyotp.TOTP(row["secret"]).verify(code, valid_window=1)
|
|
elif row.get("recovery_codes") and code in (row["recovery_codes"] or []):
|
|
valid = True
|
|
if not valid:
|
|
raise HTTPException(401, "Neispravan kod")
|
|
db_exec("DELETE FROM pgz_sport.user_2fa WHERE user_id=%s", (user["id"],))
|
|
ip, ua = _client(request)
|
|
audit(user["id"], "2fa.disable", ip=ip, ua=ua)
|
|
return {"status": "ok", "enabled": False}
|
|
|
|
@router.get("/2fa/status")
|
|
def twofa_status(user = Depends(require_user)):
|
|
row = db_one("SELECT enabled, verified_at, created_at FROM pgz_sport.user_2fa WHERE user_id=%s",
|
|
(user["id"],))
|
|
return {"enabled": bool(row and row.get("enabled")),
|
|
"configured": bool(row),
|
|
"verified_at": row.get("verified_at") if row else None}
|