# retoor <retoor@molodetz.nl>
import asyncio
import base64
import binascii
import hashlib
import html
import json
import re
import secrets
import logging
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from passlib.hash import pbkdf2_sha256
from fastapi import Request, HTTPException, status
from devplacepy.cache import TTLCache
from devplacepy.database import (
get_table,
get_user_stars,
bump_cache_version,
sync_local_cache,
notification_enabled,
)
from devplacepy.config import SESSION_MAX_AGE
logger = logging.getLogger(__name__)
def hash_password(password: str) -> str:
return pbkdf2_sha256.hash(password)
def verify_password(password: str, hashed: str) -> bool:
return pbkdf2_sha256.verify(password, hashed)
def create_session(user_uid: str, max_age_seconds: int = SESSION_MAX_AGE) -> str:
token = secrets.token_hex(32)
expires_at = datetime.now(timezone.utc) + timedelta(seconds=max_age_seconds)
sessions = get_table("sessions")
sessions.insert(
{
"session_token": token,
"user_uid": user_uid,
"created_at": datetime.now(timezone.utc).isoformat(),
"expires_at": expires_at.isoformat(),
"deleted_at": None,
"deleted_by": None,
}
)
return token
_user_cache = TTLCache(ttl=300, max_size=2000)
def _sync_auth_cache() -> None:
sync_local_cache("auth", _user_cache)
def clear_user_cache(user_uid: str) -> None:
for token, user in _user_cache.items():
if user.get("uid") == user_uid:
_user_cache.pop(token)
bump_cache_version("auth")
def clear_session_cache(token: str) -> None:
_user_cache.pop(token)
bump_cache_version("auth")
_UNSET = object()
def _user_from_session(request: Request):
token = request.cookies.get("session")
if not token:
return None
_sync_auth_cache()
cached = _user_cache.get(token)
if cached is not None:
return cached
sessions = get_table("sessions")
session = sessions.find_one(session_token=token, deleted_at=None)
if not session:
return None
expires = datetime.fromisoformat(session["expires_at"])
if expires.tzinfo is None:
expires = expires.replace(tzinfo=timezone.utc)
if expires < datetime.now(timezone.utc):
sessions.delete(id=session["id"])
return None
users = get_table("users")
user = users.find_one(uid=session["user_uid"])
if user and not user.get("is_active", True):
sessions.delete(id=session["id"])
_user_cache.pop(token)
return None
if user:
_user_cache.set(token, user)
return user
def _user_from_api_key(key: str):
if not key:
return None
_sync_auth_cache()
cache_key = f"k:{key}"
cached = _user_cache.get(cache_key)
if cached is not None:
return cached
user = get_table("users").find_one(api_key=key)
if not user or not user.get("is_active", True):
return None
_user_cache.set(cache_key, user)
return user
def _user_from_basic(header: str):
try:
raw = base64.b64decode(header.strip(), validate=True).decode("utf-8")
except (ValueError, binascii.Error, UnicodeDecodeError):
return None
if ":" not in raw:
return None
identifier, password = raw.split(":", 1)
identifier = identifier.strip()
if not identifier or not password:
return None
_sync_auth_cache()
cache_key = "b:" + hashlib.sha256(header.strip().encode()).hexdigest()
cached = _user_cache.get(cache_key)
if cached is not None:
return cached
users = get_table("users")
user = users.find_one(username=identifier) or users.find_one(
email=identifier.lower()
)
if not user or not user.get("is_active", True):
return None
if not verify_password(password, user.get("password_hash", "")):
return None
_user_cache.set(cache_key, user)
return user
def _request_has_auth(request: Request) -> bool:
return bool(
request.headers.get("X-API-KEY") or request.headers.get("Authorization")
)
def _resolve_user(request: Request):
user = _user_from_session(request)
if user:
return user
api_key = request.headers.get("X-API-KEY")
if api_key:
user = _user_from_api_key(api_key.strip())
if user:
return user
authorization = request.headers.get("Authorization", "")
scheme, _, credentials = authorization.partition(" ")
scheme = scheme.lower()
credentials = credentials.strip()
if scheme == "bearer" and credentials:
user = _user_from_api_key(credentials)
if user:
return user
if scheme == "basic" and credentials:
user = _user_from_basic(credentials)
if user:
return user
return None
def get_current_user(request: Request):
cached = getattr(request.state, "_auth_user", _UNSET)
if cached is not _UNSET:
return cached
user = _resolve_user(request)
try:
request.state._auth_user = user
except Exception:
pass
return user
def safe_next(value, default="/feed"):
if value and value.startswith("/") and not value.startswith("//"):
return value
return default
def require_user(request: Request):
user = get_current_user(request)
if not user:
from devplacepy.responses import wants_json
from devplacepy.services.audit import record as audit
audit.record(
request,
"security.authz.denied",
user=None,
result="denied",
summary=f"guest denied access to {request.url.path}",
)
if _request_has_auth(request) or wants_json(request):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
)
raise HTTPException(
status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/auth/login"}
)
return user
def is_admin(user: dict | None) -> bool:
return bool(user) and user.get("role") == "Admin"
def require_admin(request: Request):
user = require_user(request)
if not is_admin(user):
from devplacepy.responses import wants_json
from devplacepy.services.audit import record as audit
audit.record(
request,
"security.authz.denied",
user=user,
result="denied",
summary=f"non-admin denied access to {request.url.path}",
)
if wants_json(request):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Administrator access required",
)
raise HTTPException(
status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/feed"}
)
return user
def not_found(detail: str = "Not found") -> HTTPException:
return HTTPException(status_code=404, detail=detail)
def require_user_api(request: Request):
user = get_current_user(request)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required"
)
return user
def strip_html(text: str) -> str:
if not text:
return ""
text = re.sub(r"<[^>]+>", " ", text)
text = html.unescape(text)
return re.sub(r"\s+", " ", text).strip()
_FLOAT_TOKEN = "@dpfloat@"
def plain_float(value: float) -> str:
if value != value or value in (float("inf"), float("-inf")):
return json.dumps(value)
return format(Decimal(repr(value)), "f")
def _mark_floats(node):
if isinstance(node, bool):
return node
if isinstance(node, float):
return f"{_FLOAT_TOKEN}{plain_float(node)}{_FLOAT_TOKEN}"
if isinstance(node, dict):
return {key: _mark_floats(value) for key, value in node.items()}
if isinstance(node, (list, tuple)):
return [_mark_floats(item) for item in node]
return node
def pretty_json(data, indent: int = 2) -> str:
text = json.dumps(
_mark_floats(data), indent=indent, ensure_ascii=False, default=str
)
return text.replace(f'"{_FLOAT_TOKEN}', "").replace(f'{_FLOAT_TOKEN}"', "")
def slugify(text: str) -> str:
text = text.lower().strip()
text = re.sub(r"[^a-z0-9-]", "-", text)
text = re.sub(r"-+", "-", text)
return text.strip("-")
def generate_uid() -> str:
import uuid_utils
return str(uuid_utils.uuid7())
def make_combined_slug(text: str, uid: str) -> str:
short_uid = uid[:8]
slug_part = slugify(text)
if not slug_part:
return short_uid
return f"{short_uid}-{slug_part}"
def time_ago(dt_str: str) -> str:
dt = datetime.fromisoformat(dt_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
now = datetime.now(timezone.utc)
diff = now - dt
days = diff.days
if days > 30:
return dt.strftime("%d/%m/%Y")
if days > 0:
return f"{days}d ago"
hours = diff.seconds // 3600
if hours > 0:
return f"{hours}h ago"
minutes = diff.seconds // 60
if minutes > 0:
return f"{minutes}m ago"
return "just now"
def extract_mentions(content: str) -> list[str]:
if not content:
return []
return re.findall(r"(?:^|[\s(])@([a-zA-Z0-9_-]+)", content)
PUSH_ICON = "/static/apple-touch-icon.png"
DEFAULT_PUSH_URL = "/notifications"
_push_tasks: set[asyncio.Task] = set()
async def _safe_notify(user_uid: str, payload: dict[str, str]) -> None:
from devplacepy import push
try:
await push.notify_user(user_uid, payload)
except Exception as e:
logger.warning("Push delivery failed for %s: %s", user_uid, e)
def _schedule_push(user_uid: str, message: str, target_url: str | None) -> None:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
return
payload = {
"title": "DevPlace",
"message": message,
"icon": PUSH_ICON,
"url": target_url or DEFAULT_PUSH_URL,
}
task = loop.create_task(_safe_notify(user_uid, payload))
_push_tasks.add(task)
task.add_done_callback(_push_tasks.discard)
def create_notification(
user_uid: str,
notification_type: str,
message: str,
related_uid: str,
target_url: str | None = None,
) -> None:
from devplacepy.templating import clear_unread_cache
in_app = notification_enabled(user_uid, notification_type, "in_app")
push = notification_enabled(user_uid, notification_type, "push")
if in_app:
get_table("notifications").insert(
{
"uid": generate_uid(),
"user_uid": user_uid,
"type": notification_type,
"message": message,
"related_uid": related_uid,
"target_url": target_url,
"read": False,
"created_at": datetime.now(timezone.utc).isoformat(),
}
)
clear_unread_cache(user_uid)
if push:
_schedule_push(user_uid, message, target_url)
from devplacepy.services.audit import record as audit
audit.record_system(
"notification.create",
actor_kind="system",
target_type="notification",
target_uid=related_uid,
summary=f"notification created: {message}",
metadata={"type": notification_type, "in_app": in_app, "push": push},
links=[
audit.recipient(user_uid),
audit.link("source", "notification", related_uid),
],
)
def award_badge(user_uid: str, badge_name: str) -> bool:
badges = get_table("badges")
if badges.find_one(user_uid=user_uid, badge_name=badge_name):
return False
badge_uid = generate_uid()
badges.insert(
{
"uid": badge_uid,
"user_uid": user_uid,
"badge_name": badge_name,
"created_at": datetime.now(timezone.utc).isoformat(),
}
)
from devplacepy.services.audit import record as audit
audit.record_system(
"reward.badge.award",
actor_kind="user",
actor_uid=user_uid,
summary=f"user earned the {badge_name} badge",
metadata={"badge": badge_name},
links=[
audit.actor(user_uid),
audit.link("target", "badge", badge_uid, badge_name),
],
)
return True
LEVEL_XP = 100
XP_POST = 10
XP_COMMENT = 2
XP_PROJECT = 15
XP_GIST = 5
XP_UPVOTE = 5
XP_FOLLOW = 5
LEVEL_BADGES = {5: "Level 5", 10: "Level 10"}
BADGE_CATALOG = {
"Member": {"icon": "✦", "description": "Joined DevPlace"},
"First Post": {"icon": "✎", "description": "Published a first post"},
"First Comment": {"icon": "❝", "description": "Wrote a first comment"},
"First Project": {"icon": "⬢", "description": "Shared a first project"},
"First Gist": {"icon": "❡", "description": "Shared a first gist"},
"Prolific": {"icon": "✺", "description": "Published 10 posts"},
"Rising Star": {"icon": "☆", "description": "Earned 25 stars"},
"Star Author": {"icon": "★", "description": "Earned 100 stars"},
"Popular": {"icon": "◎", "description": "Reached 10 followers"},
"On Fire": {"icon": "🔥", "description": "Maintained a 7-day activity streak"},
"Level 5": {"icon": "❖", "description": "Reached level 5"},
"Level 10": {"icon": "❖", "description": "Reached level 10"},
}
def get_badge(badge_name: str) -> dict:
return BADGE_CATALOG.get(badge_name, {"icon": "✦", "description": badge_name})
def level_for_xp(xp: int) -> int:
return 1 + max(0, xp) // LEVEL_XP
def notify_badge(user_uid: str, badge_name: str) -> None:
user = get_table("users").find_one(uid=user_uid)
if not user:
return
create_notification(
user_uid,
"badge",
f"You earned the {badge_name} badge",
user_uid,
f"/profile/{user['username']}",
)
def award_xp(user_uid: str, amount: int) -> dict:
users = get_table("users")
user = users.find_one(uid=user_uid)
if not user or amount <= 0:
return {
"xp": user.get("xp", 0) if user else 0,
"level": user.get("level", 1) if user else 1,
"leveled_up": False,
}
current_xp = user.get("xp", 0) or 0
current_level = user.get("level", 1) or 1
new_xp = max(0, current_xp + amount)
new_level = level_for_xp(new_xp)
users.update({"uid": user_uid, "xp": new_xp, "level": new_level}, ["uid"])
clear_user_cache(user_uid)
leveled_up = new_level > current_level
from devplacepy.services.audit import record as audit
audit.record_system(
"reward.xp.grant",
actor_kind="user",
actor_uid=user_uid,
new_value=amount,
metadata={"amount": amount, "total_xp": new_xp},
summary=f"user earned {amount} XP",
links=[audit.actor(user_uid)],
)
if leveled_up:
audit.record_system(
"reward.level.up",
actor_kind="user",
actor_uid=user_uid,
old_value=current_level,
new_value=new_level,
metadata={"level": new_level},
summary=f"user reached level {new_level}",
links=[audit.actor(user_uid)],
)
create_notification(
user_uid,
"level",
f"You reached level {new_level}",
user_uid,
f"/profile/{user['username']}",
)
for level in range(current_level + 1, new_level + 1):
badge_name = LEVEL_BADGES.get(level)
if badge_name and award_badge(user_uid, badge_name):
notify_badge(user_uid, badge_name)
return {"xp": new_xp, "level": new_level, "leveled_up": leveled_up}
def check_milestone_badges(user_uid: str) -> list:
held = {row["badge_name"] for row in get_table("badges").find(user_uid=user_uid)}
awarded = []
if (
"Prolific" not in held
and get_table("posts").count(user_uid=user_uid) >= 10
and award_badge(user_uid, "Prolific")
):
awarded.append("Prolific")
if {"Star Author", "Rising Star"} - held:
stars = get_user_stars(user_uid)
if (
"Star Author" not in held
and stars >= 100
and award_badge(user_uid, "Star Author")
):
awarded.append("Star Author")
if (
"Rising Star" not in held
and stars >= 25
and award_badge(user_uid, "Rising Star")
):
awarded.append("Rising Star")
if (
"Popular" not in held
and get_table("follows").count(following_uid=user_uid) >= 10
and award_badge(user_uid, "Popular")
):
awarded.append("Popular")
if "On Fire" not in held:
from devplacepy.database import get_streaks
streak = get_streaks(user_uid)["current"]
if streak >= 7 and award_badge(user_uid, "On Fire"):
awarded.append("On Fire")
from devplacepy.services.audit import record as audit
audit.record_system(
"reward.streak.milestone",
actor_kind="user",
actor_uid=user_uid,
metadata={"streak": streak},
summary=f"user reached a {streak}-day streak",
links=[audit.actor(user_uid)],
)
for badge_name in awarded:
notify_badge(user_uid, badge_name)
return awarded
def award_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None:
if first_badge:
award_badge(user_uid, first_badge)
award_xp(user_uid, amount)
check_milestone_badges(user_uid)
def create_mention_notifications(content: str, actor_uid: str, target_url: str) -> None:
usernames = list(dict.fromkeys(extract_mentions(content)))
if not usernames:
return
users = get_table("users")
actor = users.find_one(uid=actor_uid)
if not actor:
return
actor_username = actor["username"]
for mentioned in users.find(users.table.columns.username.in_(usernames)):
if mentioned["uid"] != actor_uid:
create_notification(
mentioned["uid"],
"mention",
f"@{actor_username} mentioned you",
actor_uid,
target_url,
)
def format_date(dt_str: str, include_time: bool = False) -> str:
if not dt_str:
return ""
try:
dt = datetime.fromisoformat(dt_str)
if include_time:
return dt.strftime("%d/%m/%Y %H:%M")
return dt.strftime("%d/%m/%Y")
except (ValueError, TypeError):
return dt_str
Reference in New Issue View Git Blame Copy Permalink