108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
import re
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from utils import env, logger
|
|
|
|
from .. import gitea
|
|
from ..challenge import ChallengeStore, RateLimiter
|
|
from ..page import render_page
|
|
|
|
router = APIRouter(prefix="/activate")
|
|
|
|
USERNAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,39}$")
|
|
|
|
challenges = ChallengeStore()
|
|
challenge_limit = RateLimiter(env.gate.challenges_per_minute, 60)
|
|
activation_limit = RateLimiter(env.gate.activations_per_hour, 3600)
|
|
|
|
UNKNOWN = "No inactive account with that username. Check the spelling, or just sign in."
|
|
|
|
|
|
class SubmitIn(BaseModel):
|
|
username: str = Field(max_length=40)
|
|
nonce: str = Field(max_length=64)
|
|
|
|
|
|
def client_ip(request: Request) -> str:
|
|
forwarded = request.headers.get("x-forwarded-for", "")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def fail(message: str, status: int = 200) -> JSONResponse:
|
|
return JSONResponse({"ok": False, "message": message}, status_code=status)
|
|
|
|
|
|
@router.get("", response_class=HTMLResponse)
|
|
async def page() -> HTMLResponse:
|
|
return HTMLResponse(render_page())
|
|
|
|
|
|
@router.post("/challenge")
|
|
async def challenge(request: Request) -> JSONResponse:
|
|
ip = client_ip(request)
|
|
if not challenge_limit.allow(ip):
|
|
return fail("Too many attempts, wait a minute.", status=429)
|
|
return JSONResponse({"nonce": challenges.issue(ip), "ttl": env.gate.challenge_ttl})
|
|
|
|
|
|
@router.post("/submit")
|
|
async def submit(request: Request, payload: SubmitIn) -> JSONResponse:
|
|
ip = client_ip(request)
|
|
username = payload.username.strip()
|
|
|
|
if not activation_limit.allow(ip):
|
|
logger.warning(f"rate limited ip={ip}")
|
|
return fail("Too many attempts, try again later.", status=429)
|
|
|
|
if not USERNAME_RE.match(username):
|
|
return fail("That doesn't look like a username.")
|
|
|
|
refusal = challenges.redeem(payload.nonce, ip)
|
|
if refusal is not None:
|
|
logger.warning(
|
|
f"challenge rejected reason={refusal} ip={ip} username={username}"
|
|
)
|
|
return fail("Session expired, reload the page and try again.")
|
|
|
|
try:
|
|
user = await gitea.find_user(username)
|
|
except httpx.HTTPError as error:
|
|
logger.error(f"lookup failed username={username} error={error}")
|
|
return fail("Gitea is not answering right now, try again in a minute.")
|
|
|
|
if user is None or user.get("is_active"):
|
|
logger.info(f"activation miss ip={ip} username={username}")
|
|
return fail(UNKNOWN)
|
|
|
|
if user.get("is_admin") or user.get("prohibit_login"):
|
|
logger.warning(
|
|
f"activation refused for flagged account username={username} ip={ip}"
|
|
)
|
|
return fail("This account can only be activated by an administrator.")
|
|
|
|
created = datetime.fromisoformat(user["created"]).astimezone(UTC)
|
|
age_days = (datetime.now(UTC) - created).days
|
|
if age_days > env.gate.max_account_age_days:
|
|
logger.warning(
|
|
f"activation refused, too old username={username} age_days={age_days}"
|
|
)
|
|
return fail(
|
|
"This account is too old to self-activate. Contact the administrator."
|
|
)
|
|
|
|
try:
|
|
await gitea.activate_user(user)
|
|
except httpx.HTTPError as error:
|
|
logger.error(f"activation failed username={username} error={error}")
|
|
return fail("Could not activate the account, try again in a minute.")
|
|
|
|
logger.info(f"activated username={username} ip={ip} age_days={age_days}")
|
|
return JSONResponse({"ok": True, "message": "Account activated."})
|