|
# retoor <retoor@molodetz.nl>
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from devplacepy.config import SECONDS_PER_DAY
|
|
from devplacepy.database import get_table, get_int_setting
|
|
from devplacepy.utils import generate_uid
|
|
from devplacepy.services.devrant.ids import as_int, now_unix
|
|
|
|
TOKEN_KEY_BYTES = 20
|
|
|
|
|
|
def issue_token(user: dict) -> dict:
|
|
tokens = get_table("devrant_tokens")
|
|
key = secrets.token_hex(TOKEN_KEY_BYTES)
|
|
max_age = max(1, get_int_setting("session_max_age_days", 7)) * SECONDS_PER_DAY
|
|
moment = datetime.now(timezone.utc)
|
|
expire_time = int(moment.timestamp()) + max_age
|
|
user_id = as_int(user.get("id"))
|
|
token_id = tokens.insert(
|
|
{
|
|
"uid": generate_uid(),
|
|
"key": key,
|
|
"user_uid": user["uid"],
|
|
"user_id": user_id,
|
|
"expire_time": expire_time,
|
|
"created_at": moment.isoformat(),
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
return {
|
|
"id": int(token_id),
|
|
"key": key,
|
|
"expire_time": expire_time,
|
|
"user_id": user_id,
|
|
}
|
|
|
|
|
|
def resolve_user(params: dict) -> Optional[dict]:
|
|
token_id = as_int(params.get("token_id"))
|
|
token_key = (params.get("token_key") or "").strip()
|
|
user_id = as_int(params.get("user_id"))
|
|
if not token_id or not token_key:
|
|
return None
|
|
token = get_table("devrant_tokens").find_one(id=token_id, deleted_at=None)
|
|
if not token or token.get("key") != token_key:
|
|
return None
|
|
if user_id and as_int(token.get("user_id")) != user_id:
|
|
return None
|
|
if as_int(token.get("expire_time")) < now_unix():
|
|
return None
|
|
user = get_table("users").find_one(uid=token.get("user_uid"))
|
|
if not user or not user.get("is_active", True):
|
|
return None
|
|
return user
|
|
|
|
|
|
def resolve_user_by_key(key: str) -> Optional[dict]:
|
|
"""Resolve a user from a DevRant token key alone (40-char hex).
|
|
|
|
Used by the main DevPlace auth chain so that DevRant tokens work as
|
|
Bearer / X-API-KEY credentials on every DevPlace endpoint.
|
|
"""
|
|
if not key:
|
|
return None
|
|
token = get_table("devrant_tokens").find_one(key=key, deleted_at=None)
|
|
if not token:
|
|
return None
|
|
expire = as_int(token.get("expire_time"))
|
|
if expire and expire < now_unix():
|
|
return None
|
|
user = get_table("users").find_one(uid=token.get("user_uid"))
|
|
if not user or not user.get("is_active", True):
|
|
return None
|
|
return user
|
|
|
|
|
|
def revoke_all(user_uid: str) -> None:
|
|
tokens = get_table("devrant_tokens")
|
|
stamp = datetime.now(timezone.utc).isoformat()
|
|
for token in tokens.find(user_uid=user_uid, deleted_at=None):
|
|
tokens.update(
|
|
{"id": token["id"], "deleted_at": stamp, "deleted_by": user_uid}, ["id"]
|
|
)
|