JiriUhlir
786d055f6b
tlačítko Přihlásit přes Gitea na login stránku zachovaný lokální login jako nouzový admin fallback GET /portal/auth/gitea/login GET /portal/auth/gitea/callback OAuth state přes session a ověření přes secrets.compare_digest výměna code za token a načtení /api/v1/user mapování Gitea uživatele do users bez ORM nové Gitea účty dostanou role = developer, existující role se zachová disabled user is_active = 0 se nepřihlásí audit eventy auth.gitea.login.started, auth.gitea.login.success, auth.gitea.login.failed bezpečné chybové hlášky bez detailů a bez zobrazení client secretu
153 lines
3.8 KiB
Python
153 lines
3.8 KiB
Python
from typing import Any
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
from passlib.context import CryptContext
|
|
|
|
from app.db.database import get_connection
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain_password: str, password_hash: str) -> bool:
|
|
if not password_hash:
|
|
return False
|
|
|
|
return pwd_context.verify(plain_password, password_hash)
|
|
|
|
|
|
def get_user_by_username(username: str) -> dict[str, Any] | None:
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE username = ?
|
|
""",
|
|
(username,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def find_or_create_oauth_user(username: str, display_name: str, email: str) -> dict[str, Any]:
|
|
username = username.strip()
|
|
display_name = (display_name or username).strip() or username
|
|
email = (email or "").strip()
|
|
|
|
con = get_connection()
|
|
row = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE username = ?
|
|
""",
|
|
(username,),
|
|
).fetchone()
|
|
|
|
if row:
|
|
con.execute(
|
|
"""
|
|
UPDATE users
|
|
SET display_name = ?,
|
|
email = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(display_name, email, row["id"]),
|
|
)
|
|
else:
|
|
con.execute(
|
|
"""
|
|
INSERT INTO users (
|
|
username,
|
|
display_name,
|
|
email,
|
|
password_hash,
|
|
role,
|
|
is_active,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES (?, ?, ?, '', 'developer', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
|
""",
|
|
(username, display_name, email),
|
|
)
|
|
|
|
con.commit()
|
|
user_id = row["id"] if row else con.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
|
|
user = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(user)
|
|
|
|
|
|
def get_user_by_id(user_id: int) -> dict[str, Any] | None:
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def authenticate_user(username: str, password: str) -> dict[str, Any] | None:
|
|
user = get_user_by_username(username.strip())
|
|
|
|
if not user or not user.get("is_active"):
|
|
return None
|
|
|
|
if not verify_password(password, user.get("password_hash") or ""):
|
|
return None
|
|
|
|
return user
|
|
|
|
|
|
def current_user(request: Request) -> dict[str, Any] | None:
|
|
user_id = request.session.get("user_id")
|
|
if not user_id:
|
|
return None
|
|
|
|
try:
|
|
user_id = int(user_id)
|
|
except (TypeError, ValueError):
|
|
request.session.pop("user_id", None)
|
|
return None
|
|
|
|
user = get_user_by_id(user_id)
|
|
if not user or not user.get("is_active"):
|
|
request.session.pop("user_id", None)
|
|
return None
|
|
|
|
return user
|
|
|
|
|
|
def require_user(request: Request) -> dict[str, Any]:
|
|
user = current_user(request)
|
|
if user:
|
|
return user
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_303_SEE_OTHER,
|
|
headers={"Location": "/portal/login"},
|
|
)
|