feat(gitea): GERP 作为 OIDC Provider 实现 Gitea 免密登录（SSO）
新增 /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 入口
This commit is contained in:
TermiNexus committed
1 parent
6c7664ae6e
commit
7b83db89d8
3 files changed
+398
-1
No files matched your search
@@ -211,8 +211,18 @@ class GiteaApiClient:
|
||||
@classmethod
|
||||
def SessionConfig(cls):
|
||||
"""返回前端展示所需的连接信息。"""
|
||||
return {
|
||||
config = {
|
||||
'BaseUrl': cls._BaseUrl(),
|
||||
'Org': cls._OrgName(),
|
||||
'OwnerRole': cls._OwnerRole(),
|
||||
}
|
||||
# 免密登录（OIDC）：跳转 Gitea 的 OAuth2 入口（<BaseUrl>/user/oauth2/<认证源名>）
|
||||
# 认证源名从 [gitea_oidc] REDIRECT_URI 推断：/user/oauth2/<name>/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
|
||||
@@ -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 回调地址，形如 <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'),
|
||||
})
|
||||
@@ -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'),
|
||||
|
||||
Reference in new issue
Block a user