Unified_Authentication_Platform/backend/app/services/hydra_service.py

import logging
import json
import base64
from typing import List, Optional, Dict, Any

import ory_hydra_client
from ory_hydra_client.api import o_auth2_api
from ory_hydra_client.models.accept_o_auth2_login_request import AcceptOAuth2LoginRequest
from ory_hydra_client.models.reject_o_auth2_request import RejectOAuth2Request
from ory_hydra_client.models.accept_o_auth2_consent_request import AcceptOAuth2ConsentRequest
import requests

from app.core.hydra_config import hydra_settings

logger = logging.getLogger(__name__)


def decode_jwt_payload(token: str) -> Optional[Dict[str, Any]]:
    """
    解析 JWT token 的 payload（不验证签名，仅用于日志记录）。
    用于打印 id_token 的内容。
    """
    try:
        # JWT 格式: header.payload.signature
        parts = token.split('.')
        if len(parts) != 3:
            return None
        
        # 解码 payload（第二部分）
        payload_b64 = parts[1]
        # 添加 padding 如果需要
        padding = len(payload_b64) % 4
        if padding:
            payload_b64 += '=' * (4 - padding)
        
        payload_bytes = base64.urlsafe_b64decode(payload_b64)
        payload_str = payload_bytes.decode('utf-8')
        return json.loads(payload_str)
    except Exception as e:
        logger.warning(f"Failed to decode JWT payload: {e}")
        return None


class HydraService:
    def __init__(self):
        configuration = ory_hydra_client.Configuration(
            host=hydra_settings.HYDRA_ADMIN_URL
        )
        self.api_client = ory_hydra_client.ApiClient(configuration)
        self.oauth2 = o_auth2_api.OAuth2Api(self.api_client)
        # Admin REST base，用于调用 /clients 等管理接口
        self.admin_base = hydra_settings.HYDRA_ADMIN_URL.rstrip("/")

    # ---------- OAuth2 Client 管理 ----------
    def create_or_update_client(
        self,
        client_id: str,
        client_secret: str,
        redirect_uris: List[str],
        client_name: str = "",
    ) -> None:
        """
        在 Hydra 中创建或更新一个 OAuth2 Client，使其 client_id 与应用 app_id 对齐。

        - 如果已存在同名 Client，则先尝试删除再重建（简单幂等实现）。
        - grant_types / scope 等使用平台推荐的默认配置。
        """
        if not redirect_uris:
            logger.warning(
                "尝试在 Hydra 创建 Client 时未提供 redirect_uris，client_id=%s",
                client_id,
            )

        payload = {
            "client_id": client_id,
            "client_secret": client_secret,
            "client_name": client_name or client_id,
            "redirect_uris": redirect_uris,
            "grant_types": ["authorization_code", "refresh_token"],
            "response_types": ["code"],
            "scope": "openid offline offline_access profile email",
            "token_endpoint_auth_method": "client_secret_basic",
        }

        # 1. 尝试删除已有的同名 Client（容错即可）
        try:
            resp_del = requests.delete(
                f"{self.admin_base}/admin/clients/{client_id}", timeout=5
            )
            if resp_del.status_code not in (200, 204, 404):
                logger.warning(
                    "删除 Hydra Client 可能失败（忽略继续创建）: status=%s, body=%s",
                    resp_del.status_code,
                    resp_del.text,
                )
        except Exception as e:
            logger.warning("删除 Hydra Client 异常（忽略）: %s", e)

        # 2. 创建新的 Client
        resp = requests.post(
            f"{self.admin_base}/admin/clients", json=payload, timeout=5
        )
        if resp.status_code not in (200, 201):
            logger.error(
                "在 Hydra 创建 Client 失败: status=%s, body=%s, url=%s, payload=%s",
                resp.status_code,
                resp.text,
                f"{self.admin_base}/admin/clients",
                payload,
            )
            raise Exception(f"创建 Hydra OIDC Client 失败: {resp.text}")

    def get_login_request(self, challenge: str):
        try:
            return self.oauth2.get_o_auth2_login_request(challenge)
        except Exception as e:
            logger.error(f"获取登录请求失败 (challenge: {challenge}): {e}")
            raise

    def accept_login_request(self, challenge: str, subject: str):
        body = AcceptOAuth2LoginRequest(
            subject=subject,
            remember=True,
            remember_for=3600,
        )
        try:
            logger.info(f"接受登录请求 (subject: {subject}, challenge: {challenge})")
            return self.oauth2.accept_o_auth2_login_request(challenge, accept_o_auth2_login_request=body)
        except Exception as e:
            logger.error(f"接受登录请求失败 (challenge: {challenge}): {e}")
            raise

    def reject_login_request(self, challenge: str, error: str, error_description: str):
        body = RejectOAuth2Request(
            error=error,
            error_description=error_description
        )
        try:
            logger.info(f"拒绝登录请求 (challenge: {challenge}, error: {error})")
            return self.oauth2.reject_o_auth2_login_request(challenge, reject_o_auth2_request=body)
        except Exception as e:
            logger.error(f"拒绝登录请求失败 (challenge: {challenge}): {e}")
            raise

    def get_consent_request(self, challenge: str):
        try:
            return self.oauth2.get_o_auth2_consent_request(challenge)
        except Exception as e:
            logger.error(f"获取同意请求失败 (challenge: {challenge}): {e}")
            raise

    def accept_consent_request(self, challenge: str, grant_scope: list, id_token_claims: dict):
        body = AcceptOAuth2ConsentRequest(
            grant_scope=grant_scope,
            grant_access_token_audience=[],
            remember=True,
            remember_for=3600,
            session={"id_token": id_token_claims}
        )
        try:
            # 详细记录注入的 claims（格式化 JSON）
            claims_json = json.dumps(id_token_claims, ensure_ascii=False, indent=2)
            logger.info(
                f"接受同意请求 (challenge: {challenge}, scope: {grant_scope})\n"
                f"注入到 id_token 的 claims:\n{claims_json}"
            )
            result = self.oauth2.accept_o_auth2_consent_request(challenge, accept_o_auth2_consent_request=body)
            
            # 如果返回结果中包含 redirect_to，尝试从 URL 参数中提取可能的 token（仅用于调试）
            # 注意：实际的 id_token 是在 token 交换时才会生成
            if hasattr(result, 'redirect_to') and result.redirect_to:
                logger.debug(f"Consent accepted, redirect_to: {result.redirect_to}")
            
            return result
        except Exception as e:
            logger.error(f"接受同意请求失败 (challenge: {challenge}): {e}")
            raise
    
    def reject_consent_request(self, challenge: str, error: str, error_description: str):
        body = RejectOAuth2Request(
            error=error,
            error_description=error_description
        )
        try:
            logger.info(f"拒绝同意请求 (challenge: {challenge}, error: {error})")
            return self.oauth2.reject_o_auth2_consent_request(challenge, reject_o_auth2_request=body)
        except Exception as e:
            logger.error(f"拒绝同意请求失败 (challenge: {challenge}): {e}")
            raise

    def get_logout_request(self, challenge: str):
        try:
            resp = requests.get(
                f"{self.admin_base}/admin/oauth2/auth/requests/logout",
                params={"logout_challenge": challenge},
                timeout=5
            )
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.error(f"获取登出请求失败 (challenge: {challenge}): {e}")
            raise

    def accept_logout_request(self, challenge: str):
        try:
            resp = requests.put(
                f"{self.admin_base}/admin/oauth2/auth/requests/logout/accept",
                params={"logout_challenge": challenge},
                timeout=5
            )
            resp.raise_for_status()
            logger.info(f"接受登出请求 (challenge: {challenge})")
            return resp.json()
        except Exception as e:
            logger.error(f"接受登出请求失败 (challenge: {challenge}): {e}")
            raise

    def log_id_token_content(self, id_token: str, context: str = ""):
        """
        解析并打印 id_token 的内容（用于调试）。
        
        Args:
            id_token: JWT 格式的 id_token
            context: 上下文信息（如 "token exchange"）
        """
        payload = decode_jwt_payload(id_token)
        if payload:
            payload_json = json.dumps(payload, ensure_ascii=False, indent=2)
            logger.info(
                f"ID Token 内容 {context}:\n{payload_json}"
            )
        else:
            logger.warning(f"无法解析 id_token {context}")


hydra_service = HydraService()