From 7b83db89d8095ed295fc1ecd595cd6ecca9ff733 Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Tue, 11 Aug 2026 16:38:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(gitea):=20GERP=20=E4=BD=9C=E4=B8=BA=20OIDC?= =?UTF-8?q?=20Provider=20=E5=AE=9E=E7=8E=B0=20Gitea=20=E5=85=8D=E5=AF=86?= =?UTF-8?q?=E7=99=BB=E5=BD=95=EF=BC=88SSO=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 /oidc/authorize|token|jwks|userinfo 与 /.well-known/openid-configuration：authorize 校验 GERP 会话自动授权，token 经 client 凭据换 RS256 id_token（sub/preferred_username/email 携带 Gitea 用户名）；SessionConfig 输出 SsoUrl 供前端跳转 Gitea OAuth2 入口 --- gvsds/apps/gerp/gitea.py | 12 +- gvsds/apps/gerp/oidc.py | 379 +++++++++++++++++++++++++++++++++++++++ gvsds/urls.py | 8 + 3 files changed, 398 insertions(+), 1 deletion(-) create mode 100644 gvsds/apps/gerp/oidc.py diff --git a/gvsds/apps/gerp/gitea.py b/gvsds/apps/gerp/gitea.py index dc43f51e..367e2cde 100644 --- a/gvsds/apps/gerp/gitea.py +++ b/gvsds/apps/gerp/gitea.py @@ -211,8 +211,18 @@ class GiteaApiClient: @classmethod def SessionConfig(cls): """返回前端展示所需的连接信息。""" - return { + config = { 'BaseUrl': cls._BaseUrl(), 'Org': cls._OrgName(), 'OwnerRole': cls._OwnerRole(), } + # 免密登录（OIDC）：跳转 Gitea 的 OAuth2 入口（/user/oauth2/<认证源名>） + # 认证源名从 [gitea_oidc] REDIRECT_URI 推断：/user/oauth2//callback + redirect_uri = app_secrets.get('gitea_oidc', 'REDIRECT_URI', '') + if redirect_uri: + tail = redirect_uri.rsplit('/user/oauth2/', 1) + if len(tail) == 2: + name = tail[1].split('/')[0].strip() + if name: + config['SsoUrl'] = f'{cls._BaseUrl()}/user/oauth2/{name}' + return config diff --git a/gvsds/apps/gerp/oidc.py b/gvsds/apps/gerp/oidc.py new file mode 100644 index 00000000..2ffba71b --- /dev/null +++ b/gvsds/apps/gerp/oidc.py @@ -0,0 +1,379 @@ +"""模块：GERP OIDC Provider（Gitea 免密登录 SSO） + +GERP 作为 OIDC Provider（IdP），Gitea 作为 OIDC Client（RP）： +用户在 GERP 已登录 → 点「免密登录 Gitea」→ 跳转 Gitea 登录页 +→ Gitea 重定向到本模块 authorize 端点 → 校验 GERP 会话自动授权 +→ 回跳 Gitea callback 带 code → Gitea 调 token 端点换 id_token +→ Gitea 校验签名（jwks_uri）后完成登录，全程无需输入 Gitea 密码。 + +端点： + /.well-known/openid-configuration discovery（OIDC 标准路径） + /oidc/authorize GET 授权端点（浏览器 302 跳转） + /oidc/token POST code 换 id_token + /oidc/jwks GET RSA 公钥（JWK Set） + +配置（secrets.ini [gitea_oidc]）： + ISSUER IdP 基础 URL，如 https://gerp.gvsds.com（无尾斜杠） + CLIENT_ID 与 Gitea 认证源中的 Client ID 一致（如 gerp） + CLIENT_SECRET 与 Gitea 认证源中的 Client Secret 一致 + REDIRECT_URI Gitea 回调地址，形如 /user/oauth2/<认证源名>/callback + PRIVATE_KEY base64 DER 编码的 RSA 私钥（首次运行自动生成并回写） + +安全要点： +- id_token 使用 RS256 签名，私钥仅存服务端 secrets.ini，公钥经 /oidc/jwks 暴露 +- authorization code 一次性、10 分钟有效，存进程内存（与邮箱 token 同模式） +- token 端点要求 client_id+client_secret Basic 认证，防止伪造换取 +- 仅允许预配置的 redirect_uri，防开放重定向 +""" +import base64 +import hashlib +import logging +import os +import threading +import time + +import jwt +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from django.http import HttpResponseRedirect +from rest_framework.permissions import AllowAny +from rest_framework.response import Response +from rest_framework.views import APIView + +from gvsds import secrets as app_secrets + +logger = logging.getLogger(__name__) + +# === 一次性授权码存储（进程内存） === +_AUTH_CODE_TTL = 600 # code 有效期 10 分钟 +_AUTH_CODE_STORE = {} # code -> {"user_uuid": str, "nonce": str|None, "exp": float} +_AUTH_CODE_LOCK = threading.Lock() + +_ID_TOKEN_TTL = 3600 # id_token 有效期 1 小时 + + +def _PurgeAuthCodes(): + """清理已过期的授权码。""" + now = time.time() + expired = [k for k, v in _AUTH_CODE_STORE.items() if v['exp'] < now] + for k in expired: + _AUTH_CODE_STORE.pop(k, None) + + +def _GetConfig(key, fallback=''): + """读取 [gitea_oidc] 配置。""" + return app_secrets.get('gitea_oidc', key, fallback) + + +def _Issuer(): + return _GetConfig('ISSUER', 'https://gerp.gvsds.com').rstrip('/') + + +def _ClientId(): + return _GetConfig('CLIENT_ID', 'gerp') + + +def _ClientSecret(): + return _GetConfig('CLIENT_SECRET', '') + + +def _RedirectUri(): + return _GetConfig('REDIRECT_URI', '') + + +def _GetPrivateKey(): + """加载 RSA 私钥；不存在则生成 2048 位密钥并回写 secrets.ini（base64 DER 单行）。""" + b64 = _GetConfig('PRIVATE_KEY', '') + if b64: + try: + der = base64.b64decode(b64) + return serialization.load_der_private_key(der, password=None) + except Exception as e: + logger.error('[OIDC] PRIVATE_KEY 解析失败: %s', e) + # 生成新密钥并原位回写 secrets.ini [gitea_oidc] PRIVATE_KEY（含空值占位时也兼容） + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + der = key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + b64 = base64.b64encode(der).decode('ascii') + try: + path = str(app_secrets._SECRETS_FILE) + with open(path, 'r', encoding='utf-8') as f: + lines = f.readlines() + # 在 [gitea_oidc] 段内原位更新或追加 PRIVATE_KEY 行（保留注释与其它段） + section_idx = None + key_idx = None + for i, line in enumerate(lines): + stripped = line.strip() + if stripped.startswith('[gitea_oidc]'): + section_idx = i + elif section_idx is not None and stripped.startswith('['): + break + elif section_idx is not None and stripped.split('=')[0].strip().upper() == 'PRIVATE_KEY': + key_idx = i + if key_idx is not None: + lines[key_idx] = f'PRIVATE_KEY = {b64}\n' + elif section_idx is not None: + lines.insert(section_idx + 1, f'PRIVATE_KEY = {b64}\n') + else: + lines.append('\n[gitea_oidc]\nPRIVATE_KEY = {b64}\n'.format(b64=b64)) + with open(path, 'w', encoding='utf-8') as f: + f.writelines(lines) + app_secrets.reload() + logger.info('[OIDC] 已生成新的 RSA 密钥并写入 secrets.ini [gitea_oidc] PRIVATE_KEY') + except Exception as e: + logger.error('[OIDC] 写回 PRIVATE_KEY 失败: %s', e) + return key + + +def _GetPublicJwk(): + """返回 RSA 公钥的 JWK 字典（用于 /oidc/jwks）。""" + key = _GetPrivateKey() + pub = key.public_key().public_numbers() + e = pub.e.to_bytes((pub.e.bit_length() + 7) // 8, 'big') + n = pub.n.to_bytes((pub.n.bit_length() + 7) // 8, 'big') + return { + 'kty': 'RSA', + 'use': 'sig', + 'alg': 'RS256', + 'kid': hashlib.sha256(n).hexdigest()[:16], + 'n': base64.urlsafe_b64encode(n).rstrip(b'=').decode('ascii'), + 'e': base64.urlsafe_b64encode(e).rstrip(b'=').decode('ascii'), + } + + +def _VerifyGerpSession(request): + """校验 GERP 登录会话（AccessToken Cookie），返回用户实例或 None。 + + 与 CustomJWTAuthentication 同源：读取 AccessToken Cookie → SimpleJWT 校验 + UserUUID claim → 查 core User。 + """ + from gvsds.auth import JWT_ACCESS_COOKIE + from gvsds.apps.core.models import User as CoreUser + from gvsdsdk.model_utils import UUIDObj + token = request.COOKIES.get(JWT_ACCESS_COOKIE) or '' + if not token: + return None + try: + from rest_framework_simplejwt.tokens import AccessToken + payload = AccessToken(token).payload + except Exception as e: + logger.warning('[OIDC] AccessToken 校验失败: %s', e) + return None + user_uuid = payload.get('UserUUID') or payload.get('user_uuid') + if not user_uuid: + return None + try: + return CoreUser.objects.get(UserUUID=UUIDObj(user_uuid).bytes) + except Exception: + return None + + +def _GetGiteaIdentity(user): + """按 GERP 用户查 Gitea 绑定（GiteaLogin + Email），返回 dict 或 None。""" + from .models import UserGiteaAccount + rec = UserGiteaAccount.objects.filter(UserUUID=user, Status=1).first() + if rec is None: + return None + return {'login': rec.GiteaLogin, 'email': rec.Email} + + +def _BuildIdToken(user, gitea, nonce=None): + """签发 id_token（RS256），claims 中携带 Gitea 用户名/邮箱供 Gitea 匹配账号。""" + now = int(time.time()) + claims = { + 'iss': _Issuer(), + 'sub': gitea['login'], + 'aud': _ClientId(), + 'exp': now + _ID_TOKEN_TTL, + 'iat': now, + 'auth_time': now, + 'preferred_username': gitea['login'], + 'email': gitea['email'], + 'email_verified': True, + 'name': getattr(user, 'UserName', '') or '', + } + if nonce: + claims['nonce'] = nonce + kid = _GetPublicJwk()['kid'] + return jwt.encode(claims, _GetPrivateKey(), algorithm='RS256', headers={'kid': kid}) + + +class OIDCDiscoveryView(APIView): + """OIDC Discovery 文档（/.well-known/openid-configuration）。""" + + permission_classes = [AllowAny] + + def get(self, request): + base = _Issuer() + return Response({ + 'issuer': base, + 'authorization_endpoint': f'{base}/oidc/authorize', + 'token_endpoint': f'{base}/oidc/token', + 'jwks_uri': f'{base}/oidc/jwks', + 'userinfo_endpoint': f'{base}/oidc/userinfo', + 'scopes_supported': ['openid', 'profile', 'email'], + 'response_types_supported': ['code'], + 'grant_types_supported': ['authorization_code'], + 'subject_types_supported': ['public'], + 'id_token_signing_alg_values_supported': ['RS256'], + 'token_endpoint_auth_methods_supported': ['client_secret_basic', 'client_secret_post'], + 'claims_supported': ['sub', 'iss', 'aud', 'exp', 'iat', 'preferred_username', + 'email', 'email_verified', 'name', 'nonce'], + 'code_challenge_methods_supported': [], + }) + + +class OIDCAuthorizeView(APIView): + """授权端点：校验 GERP 会话 → 签发一次性 code → 302 回 Gitea callback。""" + + permission_classes = [AllowAny] + + def get(self, request): + client_id = request.query_params.get('client_id', '') + redirect_uri = request.query_params.get('redirect_uri', '') + response_type = request.query_params.get('response_type', '') + state = request.query_params.get('state', '') + nonce = request.query_params.get('nonce') + scope = request.query_params.get('scope', '') + + # 参数校验 + if response_type != 'code': + return Response({'error': 'unsupported_response_type'}, status=400) + if client_id != _ClientId(): + return Response({'error': 'unauthorized_client'}, status=400) + if redirect_uri != _RedirectUri(): + logger.warning('[OIDC] 未注册的 redirect_uri: %s', redirect_uri) + return Response({'error': 'invalid_request', 'error_description': 'redirect_uri 未注册'}, status=400) + if 'openid' not in (scope or '').split(): + return Response({'error': 'invalid_scope'}, status=400) + + # 校验 GERP 会话 + user = _VerifyGerpSession(request) + if user is None: + # 未登录：跳转 GERP 登录页（登录后回到 authorize 会因无 state 丢失， + # 因此带 returnUrl 回跳；简化：登录页登录后用户再点一次免密按钮） + from django.conf import settings + login_url = getattr(settings, 'LOGIN_URL', '/#/login') + return HttpResponseRedirect(login_url) + + # 校验用户已绑定 Gitea + gitea = _GetGiteaIdentity(user) + if gitea is None: + logger.warning('[OIDC] 用户 %s 未绑定 Gitea 账号，拒绝授权', user) + return Response( + {'error': 'access_denied', 'error_description': '请先在 GERP 申请 Gitea 账号'}, + status=403, + ) + + # 签发一次性 code（用 os.urandom 避免与 gvsds/secrets 模块名冲突） + code = base64.urlsafe_b64encode(os.urandom(24)).rstrip(b'=').decode('ascii') + with _AUTH_CODE_LOCK: + _PurgeAuthCodes() + _AUTH_CODE_STORE[code] = { + 'user_uuid': str(getattr(user, 'UserUUID', '')), + 'nonce': nonce, + 'exp': time.time() + _AUTH_CODE_TTL, + } + + sep = '&' if '?' in redirect_uri else '?' + url = f'{redirect_uri}{sep}code={code}' + if state: + url += f'&state={state}' + logger.info('[OIDC] 授权成功 user=%s -> 回跳 %s', user, redirect_uri) + return HttpResponseRedirect(url) + + +class OIDCTokenView(APIView): + """Token 端点：client 凭据认证 + code 换 id_token。""" + + permission_classes = [AllowAny] + + def _CheckClientAuth(self, request): + """校验 client_id + client_secret（Basic 头或 POST body）。""" + auth = request.META.get('HTTP_AUTHORIZATION', '') + client_id, client_secret = '', '' + if auth.startswith('Basic '): + try: + decoded = base64.b64decode(auth[6:]).decode('utf-8') + client_id, _, client_secret = decoded.partition(':') + except Exception: + pass + if not client_id: + client_id = request.data.get('client_id', '') + client_secret = request.data.get('client_secret', '') + return client_id == _ClientId() and client_secret == _ClientSecret() + + def post(self, request): + if not self._CheckClientAuth(request): + return Response({'error': 'invalid_client'}, status=401) + + grant_type = request.data.get('grant_type', '') + code = request.data.get('code', '') + redirect_uri = request.data.get('redirect_uri', '') + + if grant_type != 'authorization_code': + return Response({'error': 'unsupported_grant_type'}, status=400) + if redirect_uri != _RedirectUri(): + return Response({'error': 'invalid_request', 'error_description': 'redirect_uri 不匹配'}, status=400) + + with _AUTH_CODE_LOCK: + entry = _AUTH_CODE_STORE.pop(code, None) + if entry is None or entry['exp'] < time.time(): + return Response({'error': 'invalid_grant'}, status=400) + + # 取回用户 + from gvsds.apps.core.models import User as CoreUser + from gvsdsdk.model_utils import UUIDObj + try: + user = CoreUser.objects.get(UserUUID=UUIDObj(entry['user_uuid']).bytes) + except Exception: + return Response({'error': 'invalid_grant', 'error_description': '用户不存在'}, status=400) + + gitea = _GetGiteaIdentity(user) + if gitea is None: + return Response({'error': 'invalid_grant', 'error_description': '用户未绑定 Gitea'}, status=400) + + id_token = _BuildIdToken(user, gitea, nonce=entry['nonce']) + return Response({ + 'access_token': id_token, + 'token_type': 'Bearer', + 'expires_in': _ID_TOKEN_TTL, + 'id_token': id_token, + }) + + +class OIDCJwksView(APIView): + """JWKS 端点：返回 RSA 公钥供 Gitea 校验 id_token 签名。""" + + permission_classes = [AllowAny] + + def get(self, request): + return Response({'keys': [_GetPublicJwk()]}) + + +class OIDCUserInfoView(APIView): + """UserInfo 端点（Bearer id_token/access_token）。""" + + permission_classes = [AllowAny] + + def get(self, request): + auth = request.META.get('HTTP_AUTHORIZATION', '') + if not auth.startswith('Bearer '): + return Response({'error': 'invalid_token'}, status=401) + token = auth[7:] + try: + payload = jwt.decode(token, _GetPrivateKey().public_key(), algorithms=['RS256'], + audience=_ClientId()) + except jwt.PyJWTError as e: + logger.warning('[OIDC] userinfo token 校验失败: %s', e) + return Response({'error': 'invalid_token'}, status=401) + return Response({ + 'sub': payload.get('sub'), + 'preferred_username': payload.get('preferred_username'), + 'email': payload.get('email'), + 'email_verified': payload.get('email_verified'), + 'name': payload.get('name'), + }) diff --git a/gvsds/urls.py b/gvsds/urls.py index 669f36e6..1372c7c4 100644 --- a/gvsds/urls.py +++ b/gvsds/urls.py @@ -9,6 +9,7 @@ from rest_framework_simplejwt.views import ( from gvsds.apps.core.views import SubdomainView from gvsds.apps.ugm import views as ugm_views from gvsds.apps.gerp import views as gerp_views +from gvsds.apps.gerp import oidc as gerp_oidc from gvsds.apps.gerp.views.settings_api import SettingsAPIView from gvsds.apps.crb import views as crb_views from gvsds.apps.sve import views as sve_views @@ -182,6 +183,13 @@ urlpatterns = [ path('API/Gitea/AdminList', gerp_views.GiteaAdminListView.as_view(), name='gitea_admin_list'), path('API/Gitea/Disable', gerp_views.GiteaDisableView.as_view(), name='gitea_disable'), + # Gitea 免密登录（OIDC Provider） + path('.well-known/openid-configuration', gerp_oidc.OIDCDiscoveryView.as_view(), name='oidc_discovery'), + path('oidc/authorize', gerp_oidc.OIDCAuthorizeView.as_view(), name='oidc_authorize'), + path('oidc/token', gerp_oidc.OIDCTokenView.as_view(), name='oidc_token'), + path('oidc/jwks', gerp_oidc.OIDCJwksView.as_view(), name='oidc_jwks'), + path('oidc/userinfo', gerp_oidc.OIDCUserInfoView.as_view(), name='oidc_userinfo'), + path('API/doc/doc/GetDocs', gerp_views.DocView.as_view(), name='doc_list'), path('API/doc/doc/GetDoc', gerp_views.DocView.as_view(), name='doc_get'), path('API/doc/doc/AddDoc', gerp_views.DocView.as_view(), name='doc_add'),