forked from retoor/devplacepy
retoorandClaude Sonnet 5
57087536e5
- Route Devii-driven AI gateway cost to the action/tool that triggered it instead of a blanket "internal" bucket, so per-feature AI spend is attributable. - Fix the quiz attempt review to show one previously-answered question at a time instead of all of them at once, and stop a quiz endpoint linked from the quiz flow from responding with raw JSON. - Add DB API async query result route and AI Usage Analyzer annotated source/media routes, with traversal-safe uid/path handling and matching tests. - Add Code Farm action audit logging (plant/harvest/buy-plot/upgrade/ fertilize) and related admin workspace/services/trash/gateway route and doc touch-ups. - Drop redundant docstrings from access_tokens.py per the no-comments convention. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VL9Xn57W5UR3HZbbuuzxdK
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
# 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, is_account_active
|
|
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 is_account_active(user):
|
|
return None
|
|
return user
|
|
|
|
|
|
def resolve_user_by_key(key: str) -> Optional[dict]:
|
|
# Used by the main DevPlace auth chain so devRant tokens also 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 is_account_active(user):
|
|
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"]
|
|
)
|
|
from devplacepy.utils.authcache import clear_user_cache
|
|
|
|
clear_user_cache(user_uid)
|