# app/auth/auth_bearer.py import json from typing import Annotated import requests from fastapi import Depends, HTTPException, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer import src.config as config from .model import UserBase from .util import extract_template class JWTBearer(HTTPBearer): def __init__(self, auto_error: bool = True): # Always pass auto_error=False so that HTTPBearer returns None (instead of # raising HTTP 403) when the Authorization header is absent. We raise 401 # ourselves in the else-branch below, which is the correct status for a # missing / unauthenticated credential. super(JWTBearer, self).__init__(auto_error=False) async def __call__(self, request: Request): credentials: HTTPAuthorizationCredentials = await super( JWTBearer, self ).__call__(request) if credentials: if not credentials.scheme == "Bearer": raise HTTPException( status_code=401, detail="Invalid authentication scheme." ) method = request.method if method == "OPTIONS": return path = extract_template(request.url.path, request.path_params) endpoint = f"/optimumoh{path}" user_info, message = self.verify_jwt(credentials.credentials, method, endpoint) if not user_info: if isinstance(message, dict): message = message.get("message") or message.get("detail") or "Invalid token or expired token." else: message = str(message) if message else "Invalid token or expired token." raise HTTPException( status_code=401, detail=message ) request.state.user = message from src.context import set_user_id, set_username, set_role if hasattr(message, "user_id"): set_user_id(str(message.user_id)) username = getattr(message, "username", None) or getattr(message, "name", None) if username: set_username(username) if hasattr(message, "role"): set_role(message.role) return message else: raise HTTPException(status_code=401, detail="Invalid authorization code.") def verify_jwt(self, jwtoken: str, method: str, endpoint: str): try: url_to_verify = f"{config.AUTH_SERVICE_API}/verify-token" response = requests.get( url_to_verify, headers={"Authorization": f"Bearer {jwtoken}"}, ) if not response.ok: return False, response.json() user_data = response.json() return True, UserBase(**user_data["data"]) except Exception as e: print(f"Token verification error: {str(e)}") return False, str(e) # Create dependency to get current user from request state async def get_current_user(request: Request) -> UserBase: return request.state.user async def get_token(request: Request): token = request.headers.get("Authorization") if token: return token.replace("Bearer ", "") # Menghapus prefix "Bearer " else: return request.cookies.get("access_token") # Fallback ke cookie return "" # Mengembalikan token atau None jika tidak ada async def internal_key(request: Request): token = request.headers.get("Authorization") if not token: api_key = request.headers.get("X-Internal-Key") if api_key != config.API_KEY: raise HTTPException( status_code=401, detail="Invalid Key." ) try: headers = { 'Content-Type': 'application/json' } response = requests.post( f"{config.AUTH_SERVICE_API}/sign-in", headers=headers, data=json.dumps({ "username": "ohuser", "password": "123456789" }) ) if not response.ok: print(str(response.json())) raise Exception("error auth") user_data = response.json() return user_data['data']['access_token'] except Exception as e: raise Exception(str(e)) else: try: verify_url = f"{config.AUTH_SERVICE_API}/verify-token" response = requests.get( verify_url, headers={"Authorization": f"{token}"}, ) if not response.ok: raise HTTPException( status_code=401, detail="Invalid token." ) return token.split(" ")[1] except Exception as e: print(f"Token verification error: {str(e)}") return False, str(e) import asyncio import logging from typing import Dict, Any import src.config as config log = logging.getLogger(__name__) AUTH_NOTIFY_ENDPOINT = f"{config.AUTH_SERVICE_API}/admin/notify-limit" async def notify_admin_on_rate_limit( endpoint_name: str, ip_address: str, request: Request, method: str = "POST", cooldown: int = 900, timeout: int = 5 ) -> Dict[str, Any]: """ Kirim notifikasi ke admin via be-auth service ketika rate limit terlampaui. Async version - gunakan di async context. """ payload = { "endpoint_name": f"oh/{endpoint_name}".replace("//", ""), "ip_address": ip_address, "method": method, "cooldown": cooldown, } token = request.headers.get("Authorization") headers = {"Authorization": token} if token else {} try: response = await asyncio.to_thread( requests.post, AUTH_NOTIFY_ENDPOINT, json=payload, headers=headers, timeout=timeout ) response.raise_for_status() result = response.json() log.info(f"Notifikasi admin sent | Endpoint: {endpoint_name}") return result except Exception as e: log.error(f"Error notifying admin: {str(e)}") return {"status": False, "message": str(e), "data": payload} def notify_admin_on_rate_limit_sync( endpoint_name: str, ip_address: str, request: Request, method: str = "POST", cooldown: int = 900, timeout: int = 5 ) -> Dict[str, Any]: """ Kirim notifikasi ke admin via be-auth service. Sync version - gunakan di exception handler atau sync context. RECOMMENDED untuk use case ini. """ payload = { "endpoint_name": f"oh/{endpoint_name}".replace("//", "/"), "ip_address": ip_address, "method": method, "cooldown": cooldown, } token = request.headers.get("Authorization") headers = {"Authorization": token} if token else {} try: response = requests.post(AUTH_NOTIFY_ENDPOINT, json=payload, headers=headers, timeout=timeout) response.raise_for_status() result = response.json() log.info(f"Notifikasi admin sent | Endpoint: {endpoint_name}") return result except Exception as e: log.error(f"Error notifying admin: {str(e)}") return {"status": False, "message": str(e), "data": payload} async def admin_required(request: Request): user = await get_current_user(request) if user.role != "Admin" or user.role != "Superadmin": raise HTTPException( status_code=403, detail="Invalid authorization code." ) return user Admin = Annotated[UserBase, Depends(admin_required)] CurrentUser = Annotated[UserBase, Depends(get_current_user)] Token = Annotated[str, Depends(get_token)] InternalKey = Annotated[str, Depends(internal_key)]