懒猫微服进阶心得（十六）：解密OpenID Connect，如何自主注册懒猫SSO？

我们移植应用的时候经常希望集成懒猫微服的SSO，在打包LPK上架应用的时候，可以使用官方的配置文件进行集成。但是我们今天想要刨根问底，让他变得通用一些。如果我本地的app想要集成懒猫微服的SSO，是不是也有其他的办法呢，毕竟是底层也是通用的OpenID Connect协议，于是便有了这个文章。

懒猫的SSO是如下配置：

- Issuer：https://name.heiyu.space/sys/oauth
- HTTP 监听：0.0.0.0:8000（容器内部，没有映射到宿主机端口）
- gRPC 监听：0.0.0.0:5557，开启了 reflection
- 存储：SQLite3 内存模式（:memory:），意味着重启后所有数据丢失
- Connector：使用 authproxy 类型，名为 hportal
- 没有配置任何 staticClients（OAuth client）
- 在我的网络中，IP 为 172.18.0.2
- 没有端口映射到宿主机，应该是通过反向代理（路径 /sys/oauth）访问

懒猫微服的SSO服务运行在5557 端口，通过 docker 网络访问地址是 172.18.0.2:5557。

可以先使用grpcurl来连接测试：

1	./grpcurl -plaintext 172.18.0.2:5557 list

在这里下载GRPC：
https://github.com/fullstorydev/grpcurl/releases

gRPC 是一个跨语言、高性能的远程过程调用（RPC）框架，它强依赖 HTTP/2 协议，并默认使用 Protobuf 作为二进制序列化协议。它的核心优势在于利用 HTTP/2 的多路复用和头部压缩提升了性能，通过二进制传输减少了带宽消耗，并且通过强类型的接口定义保证了跨语言调用的严谨性。”

然后使用grpcurl对懒猫微服的SSO的API进行操作，因为是OIDC，所以这一步骤主要一个Oauth的应用，也就是注册App name，client_id，client_secret，以及 redirect_uris。存储用的是内存 SQLite，容器重启后所有 OAuth token、授权码、已注册的 client 都会丢失。当然如果你想持久化的话，也可以写到systemd启动脚本让系统自启动拉起来。

1	./grpcurl -plaintext -d '{"client":{"id":"my-app","secret":"my-app-secret","redirect_uris":["http://localhost:8080/auth/callback"],"name":"My App"}}' 172.18.0.2:5557 api.Dex/CreateClient

注册之后我们把这些信息放到authlib代码里面，把SSO串起来。

仅供本地调试
下面的最小 Demo 会把 Access Token、ID Token 写入 Flask session，并通过 /profile 原样返回浏览器，目的是观察 OIDC 的完整响应。Flask 默认 Session 是客户端 Cookie，只做签名而不是加密，不适合保存原始 Token。不要把这个调试路由部署到公网；生产环境应在服务端保存必要状态，只向浏览器返回最小化的用户资料。

from flask import Flask, redirect, url_for, session, jsonify
from authlib.integrations.flask_client import OAuth
from functools import wraps
import os

app = Flask(__name__)
app.secret_key = os.urandom(24)
oauth = OAuth(app)

oauth.register(
    name='dex',
    client_id='my-app',
    client_secret='my-app-secret',
    server_metadata_url='https://aimax.heiyu.space/sys/oauth/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'},
)

def login_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if 'user' not in session:
            return redirect(url_for('login'))
        return f(*args, **kwargs)
    return decorated

@app.route('/')
def index():
    user = session.get('user')
    if user:
        return f'Hello, {user.get("email", user.get("name", "unknown"))}. <a href="/profile">Profile</a> | <a href="/logout">Logout</a>'
    return 'Welcome! Please <a href="/login">Login</a>.'

@app.route('/login')
def login():
    return oauth.dex.authorize_redirect(url_for('authorize', _external=True))

@app.route('/auth/callback')
def authorize():
    token = oauth.dex.authorize_access_token()
    session['user'] = token.get('userinfo')
    # 仅供本地调试：生产环境不要把原始 Token 放进客户端 Session
    session['token_info'] = {
        'access_token': token.get('access_token'),
        'id_token': token.get('id_token'),
        'token_type': token.get('token_type'),
        'expires_at': token.get('expires_at'),
    }
    return redirect(url_for('index'))

@app.route('/profile')
@login_required
def profile():
    # 仅供本地调试：生产环境只返回筛选后的 userinfo
    return jsonify(userinfo=session['user'], token=session.get('token_info'))

@app.route('/logout')
def logout():
    session.pop('user', None)
    return redirect(url_for('index'))

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=True)