A R R R Associates
124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta, timezone
|
|
import hashlib
|
|
import re
|
|
import secrets
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.security.jwt_tokens import utcnow
|
|
from app.core.security.passwords import hash_password
|
|
from app.core.settings import get_settings
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.iam.password_flows_models import InviteToken, PasswordResetToken
|
|
|
|
|
|
def _hash_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def validate_password_policy(password: str) -> str | None:
|
|
s = get_settings()
|
|
if len(password or "") < s.PASSWORD_MIN_LENGTH:
|
|
return f"Password must be at least {s.PASSWORD_MIN_LENGTH} characters long."
|
|
if not re.search(r"[A-Za-z]", password or ""):
|
|
return "Password must include at least one letter."
|
|
if not re.search(r"\d", password or ""):
|
|
return "Password must include at least one number."
|
|
return None
|
|
|
|
|
|
def issue_invite_token(db: Session, user: User) -> str:
|
|
plain = secrets.token_urlsafe(32)
|
|
now = utcnow()
|
|
token = InviteToken(
|
|
user_id=user.id,
|
|
token_hash=_hash_token(plain),
|
|
created_at_utc=now,
|
|
expires_at_utc=now + timedelta(hours=get_settings().INVITE_TOKEN_HOURS),
|
|
used_at_utc=None,
|
|
)
|
|
db.add(token)
|
|
user.must_change_password = True
|
|
db.commit()
|
|
return plain
|
|
|
|
|
|
def issue_password_reset_token(db: Session, user: User) -> str:
|
|
plain = secrets.token_urlsafe(32)
|
|
now = utcnow()
|
|
token = PasswordResetToken(
|
|
user_id=user.id,
|
|
token_hash=_hash_token(plain),
|
|
created_at_utc=now,
|
|
expires_at_utc=now + timedelta(hours=get_settings().PASSWORD_RESET_HOURS),
|
|
used_at_utc=None,
|
|
)
|
|
db.add(token)
|
|
db.commit()
|
|
return plain
|
|
|
|
|
|
def _validate_unused(record) -> bool:
|
|
if not record or record.used_at_utc is not None:
|
|
return False
|
|
now = utcnow()
|
|
exp = record.expires_at_utc
|
|
if getattr(exp, "tzinfo", None) is None:
|
|
exp = exp.replace(tzinfo=timezone.utc)
|
|
return exp >= now
|
|
|
|
|
|
def accept_invite(db: Session, token: str, password: str) -> User | None:
|
|
err = validate_password_policy(password)
|
|
if err:
|
|
raise ValueError(err)
|
|
record = db.execute(select(InviteToken).where(InviteToken.token_hash == _hash_token(token))).scalar_one_or_none()
|
|
if not _validate_unused(record):
|
|
return None
|
|
user = db.execute(select(User).where(User.id == record.user_id)).scalar_one_or_none()
|
|
if not user:
|
|
return None
|
|
user.password_hash = hash_password(password)
|
|
user.must_change_password = False
|
|
user.password_changed_at_utc = utcnow().replace(tzinfo=None)
|
|
user.allow_login = True
|
|
user.is_active = True
|
|
record.used_at_utc = utcnow().replace(tzinfo=None)
|
|
db.commit()
|
|
try:
|
|
from app.modules.client_identity.service import mark_identity_activated
|
|
mark_identity_activated(db, user.id)
|
|
except Exception as exc:
|
|
print(f"[CLIENT IDENTITY ACTIVATION WARNING] user={user.id} error={exc}")
|
|
return user
|
|
|
|
|
|
def reset_password_with_token(db: Session, token: str, password: str) -> User | None:
|
|
err = validate_password_policy(password)
|
|
if err:
|
|
raise ValueError(err)
|
|
record = db.execute(select(PasswordResetToken).where(PasswordResetToken.token_hash == _hash_token(token))).scalar_one_or_none()
|
|
if not _validate_unused(record):
|
|
return None
|
|
user = db.execute(select(User).where(User.id == record.user_id)).scalar_one_or_none()
|
|
if not user:
|
|
return None
|
|
user.password_hash = hash_password(password)
|
|
user.must_change_password = False
|
|
user.password_changed_at_utc = utcnow().replace(tzinfo=None)
|
|
record.used_at_utc = utcnow().replace(tzinfo=None)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
def force_change_password(db: Session, user: User, new_password: str) -> None:
|
|
err = validate_password_policy(new_password)
|
|
if err:
|
|
raise ValueError(err)
|
|
user.password_hash = hash_password(new_password)
|
|
user.must_change_password = False
|
|
user.password_changed_at_utc = utcnow().replace(tzinfo=None)
|
|
db.commit()
|