Python中Keycloak无UserStorageProvider时,如何连接Postgres外部数据库?
理清核心概念:Keycloak UserStorage SPI vs Python客户端库
首先明确:UserStorageProvider是Keycloak服务器端的Java SPI,作用是让Keycloak自身直接对接外部用户数据库,和python-keycloak客户端库完全是两个层面的东西——Python库只负责作为客户端和Keycloak服务器交互,根本不会提供这个接口。
两种可行解决方案
方案1:基于Keycloak服务器端扩展对接Postgres(推荐)
这是官方推荐的标准方式,让Keycloak直接管理外部Postgres用户,Python应用只需要和Keycloak交互:
- 用Java实现
UserStorageProvider接口,编写对接Postgres的逻辑(比如查询用户、验证密码、同步用户属性等) - 将打包好的JAR文件部署到Keycloak服务器的
providers目录,重启Keycloak - 在Keycloak控制台的目标领域中配置该用户存储提供商,关联你的Postgres数据库
- Python应用正常使用
python-keycloak库调用Keycloak的API(比如用户登录、获取用户信息、权限校验),完全不用直接操作Postgres
方案2:Python应用直接连接Postgres+Keycloak配合
如果不想修改Keycloak服务器,可让Python应用同时对接Postgres和Keycloak,分两种场景:
场景A:同步Postgres用户到Keycloak
Python应用读取Postgres中的用户数据,同步到Keycloak,后续通过Keycloak做身份认证:
import psycopg2 from keycloak import KeycloakAdmin # 连接Postgres conn = psycopg2.connect( dbname="your_db", user="your_user", password="your_pwd", host="postgres_host" ) cur = conn.cursor() cur.execute("SELECT id, username, email, password FROM users") postgres_users = cur.fetchall() # 初始化Keycloak Admin客户端 keycloak_admin = KeycloakAdmin( server_url="http://your-keycloak/auth/", username="admin", password="admin_pwd", realm_name="your_realm", verify=True ) # 同步用户到Keycloak for user_id, username, email, pwd in postgres_users: # 检查用户是否已存在 try: keycloak_admin.get_user(username) except: # 创建用户(注意:Keycloak默认会哈希密码,若Postgres存的是明文需处理) keycloak_admin.create_user({ "username": username, "email": email, "enabled": True, "credentials": [{"value": pwd, "type": "password"}] }) cur.close() conn.close()
场景B:先验证Postgres用户,再获取Keycloak令牌
Python应用先在Postgres中验证用户身份,通过后调用Keycloak接口获取令牌,用于后续资源访问:
import psycopg2 from keycloak import KeycloakOpenID # 验证Postgres用户 def validate_postgres_user(username, password): conn = psycopg2.connect( dbname="your_db", user="your_user", password="your_pwd", host="postgres_host" ) cur = conn.cursor() # 注意:Postgres应存储哈希后的密码,这里用示例逻辑(实际要匹配哈希值) cur.execute("SELECT password FROM users WHERE username = %s", (username,)) result = cur.fetchone() cur.close() conn.close() if result and result[0] == password: # 实际替换为密码哈希校验逻辑 return True return False # 验证通过后获取Keycloak令牌 if validate_postgres_user("test_user", "test_pwd"): keycloak_openid = KeycloakOpenID( server_url="http://your-keycloak/auth/", client_id="your_client", realm_name="your_realm", client_secret_key="your_client_secret" ) token = keycloak_openid.token("test_user", "test_pwd") print("Access Token:", token["access_token"]) else: print("Invalid credentials")
内容的提问来源于stack exchange,提问作者abhi1489
相关产品推荐
相关产品推荐

