Python 中的 SAML 2.0 服务提供商

SAML 2.0 Service Provider in Python

我希望在 Python 中实施基于 SAML 2.0 的服务提供商。

我的网络应用目前都是 Flask 应用。我计划制作一个 Flask blueprint/decorator,它允许我将单点登录功能放入先前存在的应用程序中。

我已经广泛研究了 python-saml,不幸的是,存在不值得解决的依赖性问题,因为我有太多预先存在的 servers/apps whos 环境不兼容。

PySAML2 看起来它可以工作,但是文档很少,而且我很难理解哪些文档可用。 Flask 应用程序中没有使用 PySAML2 的示例。

我的身份提供者是 Okta。我设置了 Okta,以便在我登录 Okta 后,我被重定向到我的应用程序。

任何人都可以提供任何关于使用 PySAML2 的建议,或者关于如何使用 SAML 2.0 最好地验证访问我的应用程序的用户的建议吗?

更新:using PySAML2 with Okta的详细解释现在在developer.okta.com.

下面是一些在 Python/Flask 中实现 SAML SP 的示例代码。此示例代码演示了几件事:

  1. 支持多个 IdP。
  2. 使用 Flask-Login 进行用户管理。
  3. 使用 "SSO URL" 作为观众限制(以简化 IdP 上的配置)。
  4. 及时配置用户("SAML JIT")
  5. 在属性语句中传递额外的用户信息。

演示的是执行 SP 发起的身份验证请求 - 我稍后会跟进。

在某些时候,我希望围绕 pysaml2 创建一个具有固执己见的默认值的包装器。

最后,与 python-saml 一样,pysaml2 库使用了 xmlsec1 二进制文件。这也可能会导致您的服务器环境出现依赖性问题。如果是这种情况,您需要考虑用 signxml 库替换 xmlsec1

下面示例中的所有内容都应该适用于以下设置:

$ virtualenv venv
$ source venv/bin/activate
$ pip install flask flask-login pysaml2

最后,您需要在 Okta 端做一些事情才能使它起作用。

首先:在 Okta 应用程序配置的 常规 选项卡中,将应用程序配置为发送 "FirstName" 和 "LastName" 属性语句。

其次:在 Okta 应用程序配置的 单点登录 选项卡中,获取 url 并将它们放入名为 example.okta.com.metadata 的文件中。您可以使用如下命令执行此操作。

$ curl [the metadata url for your Okta application] > example.okta.com.metadata

以下是您的 Python/Flask 应用程序处理 IdP 发起的 SAML 请求所需的条件:

# -*- coding: utf-8 -*-
import base64
import logging
import os
import urllib
import uuid
import zlib

from flask import Flask
from flask import redirect
from flask import request
from flask import url_for
from flask.ext.login import LoginManager
from flask.ext.login import UserMixin
from flask.ext.login import current_user
from flask.ext.login import login_required
from flask.ext.login import login_user
from saml2 import BINDING_HTTP_POST
from saml2 import BINDING_HTTP_REDIRECT
from saml2 import entity
from saml2.client import Saml2Client
from saml2.config import Config as Saml2Config

# PER APPLICATION configuration settings.
# Each SAML service that you support will have different values here.
idp_settings = {
    u'example.okta.com': {
        u"metadata": {
            "local": [u'./example.okta.com.metadata']
        }
    },
}
app = Flask(__name__)
app.secret_key = str(uuid.uuid4())  # Replace with your secret key
login_manager = LoginManager()
login_manager.setup_app(app)
logging.basicConfig(level=logging.DEBUG)
# Replace this with your own user store
user_store = {}


class User(UserMixin):
    def __init__(self, user_id):
        user = {}
        self.id = None
        self.first_name = None
        self.last_name = None
        try:
            user = user_store[user_id]
            self.id = unicode(user_id)
            self.first_name = user['first_name']
            self.last_name = user['last_name']
        except:
            pass


@login_manager.user_loader
def load_user(user_id):
    return User(user_id)


@app.route("/")
def main_page():
    return "Hello"


@app.route("/saml/sso/<idp_name>", methods=['POST'])
def idp_initiated(idp_name):
    settings = idp_settings[idp_name]
    settings['service'] = {
        'sp': {
            'endpoints': {
                'assertion_consumer_service': [
                    (request.url, BINDING_HTTP_REDIRECT),
                    (request.url, BINDING_HTTP_POST)
                ],
            },
            # Don't verify that the incoming requests originate from us via
            # the built-in cache for authn request ids in pysaml2
            'allow_unsolicited': True,
            'authn_requests_signed': False,
            'logout_requests_signed': True,
            'want_assertions_signed': True,
            'want_response_signed': False,
        },
    }

    spConfig = Saml2Config()
    spConfig.load(settings)
    spConfig.allow_unknown_attributes = True

    cli = Saml2Client(config=spConfig)
    try:
        authn_response = cli.parse_authn_request_response(
            request.form['SAMLResponse'],
            entity.BINDING_HTTP_POST)
        authn_response.get_identity()
        user_info = authn_response.get_subject()
        username = user_info.text
        valid = True
    except Exception as e:
        logging.error(e)
        valid = False
        return str(e), 401

    # "JIT provisioning"
    if username not in user_store:
        user_store[username] = {
            'first_name': authn_response.ava['FirstName'][0],
            'last_name': authn_response.ava['LastName'][0],
            }
    user = User(username)
    login_user(user)
    # TODO: If it exists, redirect to request.form['RelayState']
    return redirect(url_for('user'))


@app.route("/user")
@login_required
def user():
    msg = u"Hello {user.first_name} {user.last_name}".format(user=current_user)
    return msg


if __name__ == "__main__":
    port = int(os.environ.get('PORT', 5000))
    if port == 5000:
        app.debug = True
    app.run(host='0.0.0.0', port=port)