feat(server): add /auth/pass/activate endpoint for pass code activation

This commit is contained in:
SashegDev
2026-05-04 21:06:56 +00:00
parent bb564e6e9b
commit e347c042d5
+95
View File
@@ -643,3 +643,98 @@ async def validate_token(request: Request, credentials: HTTPAuthorizationCredent
except Exception as e:
logger.error(f"Token validation error: {e}")
raise HTTPException(400, "Invalid request")
class ActivatePassRequest(BaseModel):
pass_code: str = Field(..., min_length=3, max_length=64)
@router.post("/pass/activate")
async def activate_pass(
body: ActivatePassRequest,
current_user: dict = Depends(get_current_user),
request: Request = None,
):
"""Activate a pass code for the current user"""
ip = request.client.host if request.client else "unknown"
with get_db() as conn:
# Check if pass exists and is active
pass_row = conn.execute(
"SELECT code, owner, is_active, expires_at, max_uses, uses, activated_by FROM passes WHERE code = ?",
(body.pass_code,),
).fetchone()
if not pass_row:
raise HTTPException(404, "Проходка не найдена")
if not pass_row["is_active"]:
raise HTTPException(400, "Проходка уже использована или отозвана")
if pass_row["uses"] >= pass_row["max_uses"]:
raise HTTPException(400, "Проходка достигла лимита использований")
if pass_row["expires_at"] and pass_row["expires_at"] < time.time():
raise HTTPException(400, "Проходка истекла")
# Check if user already has an active pass
existing = conn.execute(
"SELECT 1 FROM user_passes WHERE user_id = ? AND pass_code = ?",
(current_user["id"], body.pass_code),
).fetchone()
if existing:
raise HTTPException(409, "Эта проходка уже активирована вами")
existing_pass = conn.execute("""
SELECT 1 FROM user_passes up
JOIN passes p ON up.pass_code = p.code
WHERE up.user_id = ? AND (p.expires_at IS NULL OR p.expires_at > ?)
""", (current_user["id"], time.time())).fetchone()
if existing_pass:
raise HTTPException(409, "У вас уже есть активная проходка")
now = time.time()
# Link pass to user
conn.execute(
"INSERT INTO user_passes (user_id, pass_code, activated_at) VALUES (?, ?, ?)",
(current_user["id"], body.pass_code, now),
)
# Increment usage count
conn.execute(
"UPDATE passes SET uses = uses + 1, activated_by = ?, activated_at = ? WHERE code = ?",
(current_user["id"], now, body.pass_code),
)
# Upgrade user role if they don't have a higher role
if current_user["role"] < 1:
conn.execute(
"UPDATE users SET role = 1 WHERE id = ?",
(current_user["id"],),
)
conn.commit()
log_audit(
current_user["id"],
"pass_activated",
f"Pass activated: {body.pass_code[:8]}...",
ip,
)
logger.info(
"Pass activated",
user=current_user["username"],
user_id=current_user["id"],
pass_code=body.pass_code,
ip=ip,
)
return {
"success": True,
"message": f"Проходка активирована для {current_user['username']}",
"role": 1,
}