MMD
2747f9d4bb
The PM panel (PmPanelPage -> iframe /pm/) needs the project-management platform served at /pm/ on the ERP domain. Source: org repo Aramesh/PM, branch feature/project-management-platform. Update via: git subtree pull --prefix=pm aramesh-pm feature/project-management-platform --squash
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""Local auth — scrypt password hashing + HMAC-signed session tokens.
|
|
|
|
Deliberately simple for the MVP (single role). Casdoor SSO can be added later
|
|
the same way the ERP does it.
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
import uuid
|
|
|
|
from fastapi import Depends, Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.casdoor_auth import casdoor_user_from_token, is_casdoor_token
|
|
from app.config import settings
|
|
from app.db import get_db
|
|
from app.errors import AuthError
|
|
from app.models import PMUser
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = os.urandom(16)
|
|
dk = hashlib.scrypt(password.encode(), salt=salt, n=2**14, r=8, p=1)
|
|
return f"scrypt${salt.hex()}${dk.hex()}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
_, salt_hex, dk_hex = stored.split("$")
|
|
dk = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt_hex), n=2**14, r=8, p=1)
|
|
return hmac.compare_digest(dk.hex(), dk_hex)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _sign(payload: bytes) -> str:
|
|
return hmac.new(settings.SESSION_SECRET.encode(), payload, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def create_token(user_id: uuid.UUID) -> str:
|
|
payload = json.dumps(
|
|
{"uid": str(user_id), "exp": int(time.time()) + settings.SESSION_TTL_HOURS * 3600}
|
|
).encode()
|
|
body = base64.urlsafe_b64encode(payload).rstrip(b"=")
|
|
return f"{body.decode()}.{_sign(body)}"
|
|
|
|
|
|
def verify_token(token: str) -> uuid.UUID:
|
|
try:
|
|
body_b64, sig = token.split(".", 1)
|
|
body = body_b64.encode()
|
|
if not hmac.compare_digest(_sign(body), sig):
|
|
raise ValueError("bad signature")
|
|
payload = json.loads(base64.urlsafe_b64decode(body_b64 + "=" * (-len(body_b64) % 4)))
|
|
if payload.get("exp", 0) < time.time():
|
|
raise ValueError("expired")
|
|
return uuid.UUID(payload["uid"])
|
|
except AuthError:
|
|
raise
|
|
except Exception as exc:
|
|
raise AuthError("نشست نامعتبر یا منقضی شده است") from exc
|
|
|
|
|
|
async def get_current_user(
|
|
request: Request, db: AsyncSession = Depends(get_db)
|
|
) -> PMUser:
|
|
auth = request.headers.get("authorization", "")
|
|
if not auth.lower().startswith("bearer "):
|
|
raise AuthError("احراز هویت لازم است")
|
|
token = auth.split(" ", 1)[1].strip()
|
|
if is_casdoor_token(token):
|
|
# SSO — the same Casdoor JWT the ERP issued (one login for the suite)
|
|
return await casdoor_user_from_token(token, db)
|
|
user_id = verify_token(token)
|
|
user = await db.get(PMUser, user_id)
|
|
if user is None:
|
|
raise AuthError("کاربر یافت نشد")
|
|
return user
|
|
|
|
|
|
async def authenticate(db: AsyncSession, username: str, password: str) -> PMUser | None:
|
|
user = await db.scalar(select(PMUser).where(PMUser.username == username.strip()))
|
|
if user is None or not verify_password(password, user.password_hash):
|
|
return None
|
|
return user
|