使用Python模拟IDP实现用户SSO接入腾讯云与阿里云

站长

2024年03月12日 21:34 · 阅读数 9

感谢先行者，参考：blog.51cto.com/quietguoguo…

开发语言：Python

云端SSO方式：用户SSO

起初，是为了腾讯云开启用户SSO来接入企业SSO，但是腾讯云的文档，真是一言难尽，于是就去找阿里云的文档做参考，阿里云的文档就好一些，不过，想成功接入，还是不够。于是在网上找到【使用python构建伪IdP实现SAML接入阿里云访问控制SSO】这篇文章，最终成功实现了腾讯云接入内部SSO的功能。

本文会提供阿里云与腾讯云的接入方式，总体区别不大，只有稍微有些不同，遇到的时候会说。

第一步：生成IDP密钥对

密钥对主要作用使用公钥生成 idp 元数据文档提供给 SP 做信任，然后就是使用公私钥对 IDP 响应的断言部分XML做签名。

脚本参考 generate_key.sh，使用原作者所提供的，并未做更改。

#!/bin/bash
#
# Usage: mk_keys.sh [prefix]
#
# Makes two files: private_key.pem and certificate.pem, optionally prefixed
# with the first positional argument.
#
# Thanks to http://robinelvin.wordpress.com/2009/09/04/saml-with-django/

if [[ $# -ge 1 ]] ; then
 prefix="${1}-"
else
 prefix=""
fi
private_key="${prefix}private-key.pem"
certificate="${prefix}certificate.pem"

echo "** This utility will create the OpenSSL key and certificate for the keys app."
type -P openssl &>/dev/null || {
    echo "** This utility requires openssl but it's not installed.  Aborting." >&2;
    exit 1;
}

echo "** Starting OpenSSL Interaction ------------------------------------"
openssl genrsa > "${private_key}"
openssl req -new -x509 -key "${private_key}" -out "${certificate}" -days 3650
echo "** Finished OpenSSL Interaction ------------------------------------"

echo "** These keys were created:"
ls -l -- "${private_key}" "${certificate}"
echo "** Finished."

第二步：生成IDP元数据文档

主要提供数据：idp entity id, 公钥；如下示例中：

EntityID：http://127.0.0.1:8000/sso/saml/login

公钥(太长已简化)：MIIDazCCAlOgAwIBAgIUGDDqjlsgN

<?xml version="1.0" encoding="UTF-8"?>
<EntityDescriptor xmlns="urn:oasis:names:tc:SAML:2.0:metadata" xmlns:ds="http://www.w3.org/2000/09/xmldsig#" entityID="http://127.0.0.1:8000/sso/saml/login">
    <IDPSSODescriptor WantAuthnRequestsSigned="true" protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
        <KeyDescriptor use="signing">
            <ds:KeyInfo>
                <ds:X509Data>
                    <ds:X509Certificate>MIIDazCCAlOgAwIBAgIUGDDqjlsgN</ds:X509Certificate>
                </ds:X509Data>
            </ds:KeyInfo>
        </KeyDescriptor>
        <KeyDescriptor use="encryption">
            <ds:KeyInfo>
                <ds:X509Data>
                    <ds:X509Certificate>MIIDazCCAlOgAwIBAgIUGDD</ds:X509Certificate>
                </ds:X509Data>
            </ds:KeyInfo>
        </KeyDescriptor>
        <NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</NameIDFormat>
        <SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="http://127.0.0.1:8000/sso/saml/login"/>
        <SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="http://127.0.0.1:8000/sso/saml/login"/>
    </IDPSSODescriptor>
    <Organization>
        <OrganizationName xml:lang="en">Mouse Tech</OrganizationName>
        <OrganizationDisplayName xml:lang="en">Mouse Tech</OrganizationDisplayName>
        <OrganizationURL xml:lang="en">http://127.0.0.1:8000</OrganizationURL>
    </Organization>
</EntityDescriptor>

将内容保存为.xml后缀文件，上传到腾讯云或者阿里云即可。

第三步：用Flask实现IDP响应

下面会分别给出阿里云与腾讯云的代码示例，其中不同的地方:

跳转到IDP登录的时候，腾讯云用的POST请求，阿里云采用的是GET请求且阿里云的请求需要进行解压缩。
IDP登录后的跳转腾讯云页面与方式：如您需要指定跳转腾讯云的其他页面，可使用 cloud.tencent.com/login/saml?… 形式指定，其中 xxxx 为需要指定的地址，需要做 urlencode。
IDP登录后的跳转阿里云页面与方式：如果您的IdP支持设置RelayState参数，您可以将其配置成SSO登录成功后希望跳转到的页面URL。如果不进行配置，SSO登录成功后，将会跳转到阿里云控制台首页。

用到的第三方库： lxml：xml解析 signxml：xml签名

实现企业SSO的部分，本文不做涉及，请自行实现。

阿里云接口部分

import uuid
import base64
import datetime
import zlib
import xml.etree.ElementTree
from lxml import etree
from signxml import XMLSigner
from flask import Flask, request, render_template


app = Flask(__name__)


SP_LOCATION = 'https://signin.aliyun.com/saml/SSO'
# 注意：此处为阿里云云实际SP Entity ID，具体可以参见阿里云的元数据
SP_ENTITY_ID = 'https://signin.aliyun.com/xxx/saml/SSO'
IDP_ENTITY_ID = 'http://127.0.0.1:8000/sso/saml/login'


def get_random_id() -> str:
    """
    生成随机id 作为response id ，以_开头
    """
    # It is very important that these random IDs NOT start with a number.
    random_id = '_' + uuid.uuid4().hex
    return random_id


def utcnow(delay=0) -> str:
    """
    生成yyyy-mm-ddThh:mm:ssZ 2020-03-17T10:58:07.048Z 三位毫秒
    :param delay:
    :return:
    """
    if delay == 0:
        return datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    else:
        delay_seconds = datetime.timedelta(seconds=delay)
        now = datetime.datetime.utcnow() + delay_seconds
        return now.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]+"Z"


def xml_assertion_template(in_response_to, assertion_id, username):
    """
    断言部分：签名的时候，是对断言部分进行签名
    <ds:Signature xmlns:ds="http://www.w3.org/2000/09/xmldsig#" Id="placeholder"></ds:Signature>
    使用 signxml 签名的时候，Id="placeholder" 会在该位置替换插入签名后的xml
    :param in_response_to:
    :param assertion_id:
    :param username:
    :return:
    """
    xml_str = '<saml2:Assertion xmlns:saml2="urn:oasis:names:tc:SAML:2.0:assertion" ID="' + assertion_id + '" IssueInstant="' + utcnow() + '" Version="2.0">'
    xml_str += '<saml2:Issuer>' + IDP_ENTITY_ID + '</saml2:Issuer>'
    xml_str += '<ds:Signature xmlns:ds="http://www.w3.org/2000/09/xmldsig#" Id="placeholder"></ds:Signature>'
    xml_str += '<saml2:Subject>'
    xml_str += '<saml2:NameID>' + username + '</saml2:NameID>'
    xml_str += '<saml2:SubjectConfirmation Method="urn:oasis:names:tc:SAML:2.0:cm:bearer">'
    xml_str += '<saml2:SubjectConfirmationData InResponseTo="' + in_response_to + '" NotOnOrAfter="' + utcnow(10) + '" Recipient="' + SP_LOCATION + '"/>'
    xml_str += '</saml2:SubjectConfirmation>'
    xml_str += '</saml2:Subject>'
    xml_str += '<saml2:Conditions NotBefore="' + utcnow() + '" NotOnOrAfter="' + utcnow(10) + '">'
    xml_str += '<saml2:AudienceRestriction>'
    xml_str += '<saml2:Audience>' + SP_ENTITY_ID + '</saml2:Audience>'
    xml_str += '</saml2:AudienceRestriction>'
    xml_str += '</saml2:Conditions>'
    xml_str += '<saml2:AuthnStatement AuthnInstant="' + utcnow() + '" SessionIndex="' + assertion_id + '">'
    xml_str += '<saml2:AuthnContext>'
    xml_str += '<saml2:AuthnContextClassRef>urn:oasis:names:tc:SAML:2.0:ac:classes:unspecified</saml2:AuthnContextClassRef>'
    xml_str += '</saml2:AuthnContext>'
    xml_str += '</saml2:AuthnStatement>'
    xml_str += '</saml2:Assertion>'

    return xml_str


def xml_response_template(in_response_to, response_id, signed_xml):
    xml_str = '<saml2p:Response xmlns:saml2p="urn:oasis:names:tc:SAML:2.0:protocol" Destination="' + SP_LOCATION + '" ID="' + response_id + '" InResponseTo="' + in_response_to + \
          '" IssueInstant="' + utcnow() + '" Version="2.0">'
    xml_str += '<saml2:Issuer xmlns:saml2="urn:oasis:names:tc:SAML:2.0:assertion">' + IDP_ENTITY_ID + '</saml2:Issuer>'
    xml_str += '<saml2p:Status>'
    xml_str += '<saml2p:StatusCode Value="urn:oasis:names:tc:SAML:2.0:status:Success"/>'
    xml_str += '</saml2p:Status>'
    xml_str += signed_xml
    xml_str += '</saml2p:Response>'

    return xml_str


def sign_response_xml(xml_str):
    """
    签名XML
    本处传入的签名XML是IDP响应的XML断言部分
    """
    data_to_sign = xml_str
    with open("./static/certificate.pem", "r") as f:
        cert = f.read()
    with open("./static/private-key.pem", "r") as f:
        p_key = f.read()

    root = etree.fromstring(data_to_sign)
    signed_root = XMLSigner(
        c14n_algorithm="http://www.w3.org/2001/10/xml-exc-c14n#"
    ).sign(root, key=p_key, cert=cert)
    signed_root_str = etree.tostring(signed_root)

    return signed_root_str.decode("utf-8")


@app.route('/sso/saml/login', methods=["GET"])
def sso_saml_login():
    # 解析获取请求ID，该请求ID需要在返回的时候提供，保证请求与响应对应
    saml_request_data = request.args.get('SAMLRequest')
    saml_request_data = base64.b64decode(saml_request_data)
    idp_metadata = zlib.decompress(saml_request_data, -zlib.MAX_WBITS)
    xml_root = xml.etree.ElementTree.fromstring(idp_metadata.decode('utf-8'))
    in_response_to_id = xml_root.attrib.get("ID")
    
    # username 改为阿里云实际账号
    username = "mytest@xxx.onaliyun.com"
    response_id = get_random_id()
    assertion_id = get_random_id()

    assertion_xml = xml_assertion_template(
        in_response_to=in_response_to_id,
        assertion_id=assertion_id,
        username=username
    )
    assertion_signed_xml = sign_response_xml(assertion_xml)

    resp_xml = xml_response_template(
        in_response_to=in_response_to_id,
        response_id=response_id,
        signed_xml=assertion_signed_xml
    )
    resp_b64 = base64.b64encode(resp_xml.encode("utf-8"))

    return render_template(
        "response_saml_aliyun.html",
        saml_response=resp_b64.decode("utf-8"),
        relay_state="https%3A%2F%2Fram.console.aliyun.com%2Foverview"
    )


if __name__ == '__main__':
    app.run(port=8000, debug=True)

腾讯云接口部分

import uuid
import base64
import datetime
import zlib
import xml.etree.ElementTree
from lxml import etree
from signxml import XMLSigner
from flask import Flask, request, render_template


app = Flask(__name__)


SP_LOCATION = 'https://cloud.tencent.com/saml/sso'
# 注意：此处为腾讯云实际SP Entity ID，具体可以参见腾讯云的元数据
SP_ENTITY_ID = 'https://cloud.tencent.com/xxx/saml/sso'
IDP_ENTITY_ID = 'http://127.0.0.1:8000/sso/saml/login'


def get_random_id() -> str:
    """
    生成随机id 作为response id ，以_开头
    """
    # It is very important that these random IDs NOT start with a number.
    random_id = '_' + uuid.uuid4().hex
    return random_id


def utcnow(delay=0) -> str:
    """
    生成yyyy-mm-ddThh:mm:ssZ 2020-03-17T10:58:07.048Z 三位毫秒
    :param delay:
    :return:
    """
    if delay == 0:
        return datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    else:
        delay_seconds = datetime.timedelta(seconds=delay)
        now = datetime.datetime.utcnow() + delay_seconds
        return now.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]+"Z"


def xml_assertion_template(in_response_to, assertion_id, username):
    """
    断言部分：签名的时候，是对断言部分进行签名
    <ds:Signature xmlns:ds="http://www.w3.org/2000/09/xmldsig#" Id="placeholder"></ds:Signature>
    使用 signxml 签名的时候，Id="placeholder" 会在该位置替换插入签名后的xml
    :param in_response_to:
    :param assertion_id:
    :param username:
    :return:
    """
    xml_str = '<saml2:Assertion xmlns:saml2="urn:oasis:names:tc:SAML:2.0:assertion" ID="' + assertion_id + '" IssueInstant="' + utcnow() + '" Version="2.0">'
    xml_str += '<saml2:Issuer>' + IDP_ENTITY_ID + '</saml2:Issuer>'
    xml_str += '<ds:Signature xmlns:ds="http://www.w3.org/2000/09/xmldsig#" Id="placeholder"></ds:Signature>'
    xml_str += '<saml2:Subject>'
    xml_str += '<saml2:NameID>' + username + '</saml2:NameID>'
    xml_str += '<saml2:SubjectConfirmation Method="urn:oasis:names:tc:SAML:2.0:cm:bearer">'
    xml_str += '<saml2:SubjectConfirmationData InResponseTo="' + in_response_to + '" NotOnOrAfter="' + utcnow(10) + '" Recipient="' + SP_LOCATION + '"/>'
    xml_str += '</saml2:SubjectConfirmation>'
    xml_str += '</saml2:Subject>'
    xml_str += '<saml2:Conditions NotBefore="' + utcnow() + '" NotOnOrAfter="' + utcnow(10) + '">'
    xml_str += '<saml2:AudienceRestriction>'
    xml_str += '<saml2:Audience>' + SP_ENTITY_ID + '</saml2:Audience>'
    xml_str += '</saml2:AudienceRestriction>'
    xml_str += '</saml2:Conditions>'
    xml_str += '<saml2:AuthnStatement AuthnInstant="' + utcnow() + '" SessionIndex="' + assertion_id + '">'
    xml_str += '<saml2:AuthnContext>'
    xml_str += '<saml2:AuthnContextClassRef>urn:oasis:names:tc:SAML:2.0:ac:classes:unspecified</saml2:AuthnContextClassRef>'
    xml_str += '</saml2:AuthnContext>'
    xml_str += '</saml2:AuthnStatement>'
    xml_str += '</saml2:Assertion>'

    return xml_str


def xml_response_template(in_response_to, response_id, signed_xml):
    xml_str = '<saml2p:Response xmlns:saml2p="urn:oasis:names:tc:SAML:2.0:protocol" Destination="' + SP_LOCATION + '" ID="' + response_id + '" InResponseTo="' + in_response_to + \
          '" IssueInstant="' + utcnow() + '" Version="2.0">'
    xml_str += '<saml2:Issuer xmlns:saml2="urn:oasis:names:tc:SAML:2.0:assertion">' + IDP_ENTITY_ID + '</saml2:Issuer>'
    xml_str += '<saml2p:Status>'
    xml_str += '<saml2p:StatusCode Value="urn:oasis:names:tc:SAML:2.0:status:Success"/>'
    xml_str += '</saml2p:Status>'
    xml_str += signed_xml
    xml_str += '</saml2p:Response>'

    return xml_str


def sign_response_xml(xml_str):
    """
    签名XML
    本处传入的签名XML是IDP响应的XML断言部分
    """
    data_to_sign = xml_str
    with open("./static/certificate.pem", "r") as f:
        cert = f.read()
    with open("./static/private-key.pem", "r") as f:
        p_key = f.read()

    root = etree.fromstring(data_to_sign)
    signed_root = XMLSigner(
        c14n_algorithm="http://www.w3.org/2001/10/xml-exc-c14n#"
    ).sign(root, key=p_key, cert=cert)
    signed_root_str = etree.tostring(signed_root)

    return signed_root_str.decode("utf-8")


@app.route('/sso/saml/login', methods=["POST"])
def sso_saml_login():
    # 解析获取请求ID，该请求ID需要在返回的时候提供，保证请求与响应对应
    saml_request_data = request.form.get('SAMLRequest')
    saml_request_data = base64.b64decode(saml_request_data)
    idp_metadata = saml_request_data
    xml_root = xml.etree.ElementTree.fromstring(idp_metadata.decode('utf-8'))
    in_response_to_id = xml_root.attrib.get("ID")

    username = "mytest"
    response_id = get_random_id()
    assertion_id = get_random_id()

    assertion_xml = xml_assertion_template(
        in_response_to=in_response_to_id,
        assertion_id=assertion_id,
        username=username
    )
    assertion_signed_xml = sign_response_xml(assertion_xml)

    resp_xml = xml_response_template(
        in_response_to=in_response_to_id,
        response_id=response_id,
        signed_xml=assertion_signed_xml
    )
    resp_b64 = base64.b64encode(resp_xml.encode("utf-8"))

    return render_template(
        "response_saml_tencent.html",
        saml_response=resp_b64.decode("utf-8"),
        s_url="https%3A%2F%2Fconsole.cloud.tencent.com%2F"
    )


if __name__ == '__main__':
    app.run(port=8000, debug=True)

用于响应跳转的html模板

主要区别：登录后跳转到云端指定页面的方式不一样。

IDP登录后的跳转腾讯云页面与方式：如您需要指定跳转腾讯云的其他页面，可使用 cloud.tencent.com/login/saml?… 形式指定，其中 xxxx 为需要指定的地址，需要做 urlencode。
IDP登录后的跳转阿里云页面与方式：如果您的IdP支持设置RelayState参数，您可以将其配置成SSO登录成功后希望跳转到的页面URL。如果不进行配置，SSO登录成功后，将会跳转到阿里云控制台首页。

阿里云

<!DOCTYPE html>
<html lang="en">
<head>
</head>
<body>
<p>登录跳转中，请稍后</p>
 <form id="saml" method="post" action="https://signin.aliyun.com/saml/SSO">
  <input type="hidden" name="SAMLResponse" value="{{ saml_response }}" />
  <input type="hidden" name="RelayState" value="{{ relay_state }}" />
 </form>
</body>
</html>

<script>
var frm = document.getElementById('saml');
frm.submit();
</script>

腾讯云

<!DOCTYPE html>
<html lang="en">
<head>
</head>
<body>
<p>登录跳转中，请稍后</p>
 <form id="saml" method="post" action="https://cloud.tencent.com/saml/sso?s_url={{ s_url }}">
  <input type="hidden" name="SAMLResponse" value="{{ saml_response }}" />
 </form>
</body>
</html>

<script>
var frm = document.getElementById('saml');
frm.submit();
</script>

如果有需要，就试试吧。

转载自:https://juejin.cn/post/7321896126312005647