auth.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361
  1. from fastapi import APIRouter, Request, Depends, HTTPException, WebSocket, WebSocketDisconnect, Query, Body
  2. from typing import Optional, List
  3. from services.global_manager import global_manager
  4. from services.rate_limit_service import rate_limit_service
  5. import auth_utils
  6. import db
  7. import schemas
  8. import session_utils
  9. import uuid
  10. from datetime import datetime, timedelta
  11. import locales
  12. from dependencies import get_current_user, require_admin
  13. import config
  14. import secrets
  15. from services.email_service import send_verification_email, send_password_reset_email
  16. from services.token_service import token_service
  17. from services.audit_service import audit_service
  18. try:
  19. from google.oauth2 import id_token
  20. from google.auth.transport import requests as google_requests
  21. except ImportError:
  22. id_token = None
  23. google_requests = None
  24. router = APIRouter(prefix="/auth", tags=["auth"])
  25. @router.post("/register", response_model=schemas.UserResponse)
  26. async def register(request: Request, user: schemas.UserCreate, lang: str = "en"):
  27. existing_user = db.execute_query("SELECT id FROM users WHERE email = %s", (user.email,))
  28. if existing_user:
  29. raise HTTPException(status_code=400, detail=locales.translate_error("email_already_registered", lang))
  30. ip_address = request.client.host if request.client else None
  31. hashed_password = auth_utils.get_password_hash(user.password)
  32. query = """
  33. INSERT INTO users (email, password_hash, first_name, last_name, phone, shipping_address, preferred_language, role, ip_address, is_company, company_name, company_pib, company_address, is_active)
  34. VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 0)
  35. """
  36. params = (user.email, hashed_password, user.first_name, user.last_name, user.phone, user.shipping_address, user.preferred_language, 'user', ip_address, user.is_company, user.company_name, user.company_pib, user.company_address)
  37. user_id = db.execute_commit(query, params)
  38. # Generate Verification Token (Redis)
  39. token = token_service.create_verification_token(user_id)
  40. # Send Email
  41. send_verification_email(user.email, token, user.preferred_language or lang)
  42. new_user = db.execute_query("SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users WHERE id = %s", (user_id,))
  43. await audit_service.log(
  44. user_id=user_id,
  45. action="user_register",
  46. target_type="user",
  47. target_id=user_id,
  48. details={"email": user.email, "ip": ip_address},
  49. request=request
  50. )
  51. return new_user[0]
  52. @router.get("/verify-email")
  53. async def verify_email(token: str, lang: str = "en"):
  54. user_id = token_service.verify_email_token(token)
  55. if not user_id:
  56. raise HTTPException(status_code=400, detail="Invalid or expired verification token")
  57. db.execute_commit("UPDATE users SET is_active = 1 WHERE id = %s", (user_id,))
  58. token_service.delete_verification_token(token)
  59. return {"message": "Email verified successfully. You can now log in."}
  60. @router.post("/login", response_model=schemas.Token)
  61. async def login(request: Request, user_data: schemas.UserLogin, lang: str = "en"):
  62. ip = request.client.host if request.client else "unknown"
  63. email = user_data.email.lower()
  64. # 1. Check Global Rate Limit
  65. if rate_limit_service.is_rate_limited(email, ip):
  66. raise HTTPException(
  67. status_code=429,
  68. detail=locales.translate_error("too_many_attempts", lang)
  69. )
  70. # 2. Check if Captcha is Required
  71. if rate_limit_service.is_captcha_required(email, ip):
  72. if not user_data.captcha_token:
  73. raise HTTPException(
  74. status_code=403,
  75. detail=locales.translate_error("captcha_required", lang)
  76. )
  77. # 3. Verify Captcha
  78. if not await rate_limit_service.verify_captcha(user_data.captcha_token):
  79. raise HTTPException(
  80. status_code=403,
  81. detail=locales.translate_error("invalid_token", lang)
  82. )
  83. # 4. Attempt Authentication
  84. user = db.execute_query("SELECT * FROM users WHERE email = %s", (email,))
  85. if not user or not auth_utils.verify_password(user_data.password, user[0]['password_hash']):
  86. # Log failure
  87. rate_limit_service.record_failed_attempt(email, ip)
  88. raise HTTPException(status_code=401, detail=locales.translate_error("incorrect_credentials", lang))
  89. if not user[0].get('is_active', True):
  90. raise HTTPException(status_code=403, detail=locales.translate_error("account_not_active", lang))
  91. # 5. Success - Reset Rate Limits
  92. rate_limit_service.reset_attempts(email, ip)
  93. access_token = auth_utils.create_access_token(
  94. data={"sub": user[0]['email'], "id": user[0]['id'], "role": user[0]['role']}
  95. )
  96. await audit_service.log(
  97. user_id=user[0]['id'],
  98. action="user_login",
  99. target_type="user",
  100. target_id=user[0]['id'],
  101. details={"ip": ip, "method": "credentials"},
  102. request=request
  103. )
  104. return {"access_token": access_token, "token_type": "bearer"}
  105. @router.post("/social-login", response_model=schemas.Token)
  106. async def social_login(request: Request, data: schemas.SocialLogin):
  107. email = data.email.lower() if data.email else None
  108. first_name = data.first_name
  109. last_name = data.last_name
  110. # 1. Verify token if provider is Google
  111. if data.provider == 'google':
  112. print(f"DEBUG: Social Login attempt. id_token library available: {id_token is not None}")
  113. print(f"DEBUG: config.GOOGLE_CLIENT_ID exists: {bool(config.GOOGLE_CLIENT_ID)}")
  114. print(f"DEBUG: config.GOOGLE_CLIENT_ID value: {config.GOOGLE_CLIENT_ID}")
  115. if not id_token or not config.GOOGLE_CLIENT_ID:
  116. msg = f"Config error: id_token_lib={id_token is not None}, client_id_set={bool(config.GOOGLE_CLIENT_ID)}"
  117. raise HTTPException(status_code=500, detail=f"Google Auth not configured on server ({msg})")
  118. try:
  119. # Verify the ID token
  120. idinfo = id_token.verify_oauth2_token(data.token, google_requests.Request(), config.GOOGLE_CLIENT_ID)
  121. # ID token is valid. Get user's Google info
  122. if idinfo['iss'] not in ['accounts.google.com', 'https://accounts.google.com']:
  123. raise ValueError('Wrong issuer.')
  124. email = idinfo['email'].lower()
  125. first_name = idinfo.get('given_name', first_name)
  126. last_name = idinfo.get('family_name', last_name)
  127. except Exception as e:
  128. print(f"Google Token Verification Error: {e}")
  129. raise HTTPException(status_code=401, detail="Invalid Google token")
  130. if not email:
  131. raise HTTPException(status_code=400, detail="Email is required")
  132. # 2. Proceed with login/registration
  133. user = db.execute_query("SELECT id, email, role, is_active FROM users WHERE email = %s", (email,))
  134. if user:
  135. if not user[0].get('is_active', True):
  136. raise HTTPException(status_code=403, detail="Your account has been suspended.")
  137. access_token = auth_utils.create_access_token(
  138. data={"sub": user[0]['email'], "id": user[0]['id'], "role": user[0]['role']}
  139. )
  140. return {"access_token": access_token, "token_type": "bearer"}
  141. else:
  142. ip_address = request.client.host if request.client else None
  143. hashed_password = auth_utils.get_password_hash(str(uuid.uuid4()))
  144. query = """
  145. INSERT INTO users (email, password_hash, first_name, last_name, preferred_language, role, ip_address, is_active)
  146. VALUES (%s, %s, %s, %s, %s, %s, %s, 1)
  147. """
  148. params = (email, hashed_password, first_name, last_name, data.preferred_language, 'user', ip_address)
  149. user_id = db.execute_commit(query, params)
  150. access_token = auth_utils.create_access_token(data={"sub": email, "id": user_id, "role": 'user'})
  151. return {"access_token": access_token, "token_type": "bearer"}
  152. @router.post("/logout")
  153. async def logout(user: dict = Depends(get_current_user)):
  154. sid = user.get("sid")
  155. if sid: session_utils.delete_session(sid)
  156. return {"message": "Successfully logged out"}
  157. @router.post("/forgot-password")
  158. async def forgot_password(request: schemas.ForgotPassword, lang: str = "en"):
  159. user = db.execute_query("SELECT id, preferred_language FROM users WHERE email = %s", (request.email,))
  160. if not user: raise HTTPException(status_code=404, detail="Email not found")
  161. # Generate Reset Token (Redis - 10 min TTL)
  162. token = token_service.create_reset_token(user[0]['id'])
  163. # Send Email
  164. user_lang = user[0]['preferred_language'] or lang
  165. send_password_reset_email(request.email, token, user_lang)
  166. return {"message": "Reset instructions sent to your email"}
  167. @router.post("/verify-reset-token")
  168. async def verify_reset_token_post(data: schemas.TokenVerify, lang: str = "en"):
  169. return await _verify_token_internal(data.token, lang)
  170. @router.get("/verify-reset-token")
  171. async def verify_reset_token_get(token: str, lang: str = "en"):
  172. return await _verify_token_internal(token, lang)
  173. async def _verify_token_internal(token: str, lang: str):
  174. if not token:
  175. raise HTTPException(status_code=400, detail="Token required")
  176. user_id = token_service.verify_reset_token(token)
  177. if not user_id:
  178. msg = "Invalid or expired reset token"
  179. if lang == "ru": msg = "Ссылка истекла или недействительна"
  180. elif lang == "me": msg = "Link je istekao ili je nevažeći"
  181. elif lang == "ua": msg = "Посилання закінчилося або є недійсним"
  182. raise HTTPException(status_code=400, detail=msg)
  183. return {"message": "Token is valid"}
  184. @router.post("/reset-password")
  185. async def reset_password(request: schemas.ResetPassword):
  186. user_id = token_service.verify_reset_token(request.token)
  187. if not user_id:
  188. raise HTTPException(status_code=400, detail="Invalid or expired reset token")
  189. hashed_password = auth_utils.get_password_hash(request.new_password)
  190. db.execute_commit("UPDATE users SET password_hash = %s WHERE id = %s", (hashed_password, user_id))
  191. # Successful reset - Cleanup ALL reset tokens AND sessions for this user
  192. token_service.cleanup_reset_tokens(user_id)
  193. session_utils.delete_all_user_sessions(user_id)
  194. await audit_service.log(
  195. user_id=user_id,
  196. action="password_reset_success",
  197. target_type="user",
  198. target_id=user_id
  199. )
  200. return {"message": "Password updated successfully"}
  201. @router.get("/me", response_model=schemas.UserResponse)
  202. async def get_me(user: dict = Depends(get_current_user)):
  203. user_id = user.get("id")
  204. user_data = db.execute_query("SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users WHERE id = %s", (user_id,))
  205. if not user_data: raise HTTPException(status_code=404, detail="User not found")
  206. return user_data[0]
  207. @router.put("/me", response_model=schemas.UserResponse)
  208. async def update_me(data: schemas.UserUpdate, user: dict = Depends(get_current_user)):
  209. user_id = user.get("id")
  210. update_fields = []
  211. params = []
  212. for field, value in data.dict(exclude_unset=True).items():
  213. update_fields.append(f"{field} = %s")
  214. params.append(value)
  215. if update_fields:
  216. query = f"UPDATE users SET {', '.join(update_fields)} WHERE id = %s"
  217. params.append(user_id)
  218. db.execute_commit(query, tuple(params))
  219. user = db.execute_query("SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users WHERE id = %s", (user_id,))
  220. return user[0]
  221. @router.get("/admin/users")
  222. async def admin_get_users(page: int = 1, size: int = 50, search: Optional[str] = None, admin: dict = Depends(require_admin)):
  223. offset = (page - 1) * size
  224. base_query = "SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users"
  225. count_query = "SELECT COUNT(*) as total FROM users"
  226. params = []
  227. if search and search.strip():
  228. where_clause = " WHERE email LIKE %s OR first_name LIKE %s OR last_name LIKE %s OR phone LIKE %s"
  229. base_query += where_clause
  230. count_query += where_clause
  231. pattern = f"%{search.strip()}%"
  232. params = [pattern] * 4
  233. base_query += " ORDER BY id DESC LIMIT %s OFFSET %s"
  234. users = db.execute_query(base_query, tuple(params + [size, offset]))
  235. total = db.execute_query(count_query, tuple(params))[0]['total']
  236. return {"users": users, "total": total, "page": page, "size": size}
  237. @router.post("/admin/users", response_model=schemas.UserResponse)
  238. async def admin_create_user(request: Request, data: schemas.UserCreate, admin: dict = Depends(require_admin)):
  239. existing_user = db.execute_query("SELECT id FROM users WHERE email = %s", (data.email,))
  240. if existing_user:
  241. raise HTTPException(status_code=400, detail="Email already registered")
  242. hashed_password = auth_utils.get_password_hash(data.password)
  243. user_id = db.execute_commit(
  244. "INSERT INTO users (email, password_hash, first_name, last_name, phone, role, can_chat) VALUES (%s, %s, %s, %s, %s, %s, %s)",
  245. (data.email, hashed_password, data.first_name, data.last_name, data.phone, 'user', True)
  246. )
  247. await audit_service.log(
  248. user_id=admin.get("id"),
  249. action="admin_create_user",
  250. target_type="user",
  251. target_id=user_id,
  252. details={"email": data.email, "role": "user"},
  253. request=request
  254. )
  255. user = db.execute_query("SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users WHERE id = %s", (user_id,))
  256. return user[0]
  257. @router.patch("/users/{target_id}/admin", response_model=schemas.UserResponse)
  258. async def admin_update_user(request: Request, target_id: int, data: schemas.AdminUserUpdate, admin: dict = Depends(require_admin)):
  259. update_fields = []
  260. params = []
  261. update_dict = data.dict(exclude_unset=True)
  262. # Handle password hashing
  263. if "password" in update_dict:
  264. password = update_dict.pop("password")
  265. update_dict["password_hash"] = auth_utils.get_password_hash(password)
  266. for field, value in update_dict.items():
  267. update_fields.append(f"`{field}` = %s")
  268. params.append(value)
  269. if update_fields:
  270. query = f"UPDATE users SET {', '.join(update_fields)} WHERE id = %s"
  271. params.append(target_id)
  272. db.execute_commit(query, tuple(params))
  273. await audit_service.log(
  274. user_id=admin.get("id"),
  275. action="admin_update_user",
  276. target_type="user",
  277. target_id=target_id,
  278. details={"updated_fields": {k: ('***' if k == 'password_hash' else v) for k, v in update_dict.items()}},
  279. request=request
  280. )
  281. # If user was deactivated, kick from active sessions
  282. if update_dict.get("is_active") is False:
  283. await global_manager.kick_user(target_id)
  284. user = db.execute_query("SELECT id, email, first_name, last_name, phone, shipping_address, preferred_language, role, can_chat, is_active, is_company, company_name, company_pib, company_address, ip_address, created_at FROM users WHERE id = %s", (target_id,))
  285. if not user: raise HTTPException(status_code=404, detail="User not found")
  286. return user[0]
  287. # WebSocket implementation moved to main.py to handle path prefixing issues
  288. # @router.websocket("/ws/global")
  289. # async def ws_global(websocket: WebSocket, token: str = Query(...)):
  290. # ...