Add support for OAUTH2/OpenID Connect providers to authorize internal users logins based on external corporate policies

[evilaliv3]

Proposal

This ticket is about tracking research and development to add support for OAUTH2/OpenID Connect providers.

Motivation and context

Basic idea is to use the functionality to enable support for third party authentications via OAUTH2/OpenID Connect providers to authorize to be used to implement internal users logins based on external corporate policies.

Requirement collected while working with:

• Italian National Authority for Anticorruption (ANAC) that aims at integratin GlobaLeaks with Keycloak
• Bank of Italy
• Spanish Ministry of Justice

[evilaliv3]

After looking at the available libraries ready available in open-source, written in python/typescript and available in Debian/Ubuntu i consider that a very good option would be to base the client implementation on angular-oauth2-oidc and to implement a simple backend validator of the tokens based on requests-oauthlib that uses python3-oauthlib

An advantage of angular-oauth2-oidc is also that it is openid certified so it would probably guarantee significant compatibility: https://openid.net/developers/certified-openid-connect-implementations;

In particular with such a choice we could keep the backend implementation very small reducing both the footprint and the attack surface. The code in fact could be something like:

import requests
from oauthlib.oauth2 import BackendApplicationClient
from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth2Session

def validate_oauth_token(introspection_url, client_id, client_secret, token):
    # Set up the client credentials
    client = BackendApplicationClient(client_id=client_id)
    oauth = OAuth2Session(client=client)

    # Prepare the request to the introspection endpoint
    auth = HTTPBasicAuth(client_id, client_secret)
    introspection_response = oauth.post(
        introspection_url,
        data={'token': token},
        auth=auth
    )

    if introspection_response.status_code == 200:
        token_info = introspection_response.json()
        return token_info
    else:
        # Handle the error or invalid response
        introspection_response.raise_for_status()

introspection_url = 'https://authorization-server.com/oauth/introspect'
client_id = 'your-client-id'
client_secret = 'your-client-secret'
token = 'the-token-to-validate'

token_info = validate_oauth_token(introspection_url, client_id, client_secret, token)

print(token_info)

[rglauco]

i deeply agree, creating a common interface to any IAM software (keycloak, wso2) is a valuable solution.

[mspasiano]

Now there are many open source projects on both client and server sides for integration with identity and access management, we at CNR having projects developed in Angular mainly use angular-oauth-oidc-client, while on the server side having the majority of projects in java with spring-boot we use an official spring plugin, in conclusion it is not advisable to write custom code or even use any middelware that would compromise the authorization part, but just find the plugin best suited to the GlobaLeaks architecture.

[evilaliv3]

Thank you @rglauco and @mspasiano , for your your valuable feedback.

A plugin does not exist in python and specifically for twisted, but based on the very standard and complete python3-oauthlib library we could easily verify the tokens that we receive at the first time users provide them and automatically upon a certain timeout so to invalidate the tokens upon .expiration. The code i drafted above is an example of how to handle this simple part.

[evilaliv3]

I'm going to remodel this ticket to handle all at once OATUH2 and OpenID connect because actually i think it will be more simple to keep the discussions unified and target an implementation that eventually could support both a simple OAUTH2 integration or a full OpenID connect implementation.

Regarding the server validation of the tokens i think it could be furtherly made more efficient if we could periodically fetch the signature keys from the OpenID connect server on the openid-connect/certs handler and then use them to validate tokens; this would simplify the implementation not requiring continous communication with the identity server. The idea is that globaleaks for security purposes could ask just for time limited tokens (e.g specifying the 'exp' parameter when getting the token) so to use those tokens just for authentication purposes validating the identity of users and then using regular sessions.

In this case the implementation the backend implementation can be ultimately simplified (as it would not require anyopenid connect library on the backend) but just the jose library (that we are already including due to lets'encrypt) and the code can be simplified to something like:

import json
from twisted.internet import reactor, defer
from twisted.web.client import Agent, readBody
from twisted.web.http_headers import Headers
from jose import jwt, jwk
from jose.utils import base64url_decode
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.web.http_headers import Headers

# Define the Keycloak JWKS URL
jwks_url = 'https://<keycloak-domain>/auth/realms/<realm-name>/protocol/openid-connect/certs'

# Your JWT token to verify
token = '<your-jwt-token>'

def fetch_jwks(url):
    agent = Agent(reactor)
    d = agent.request(
        b'GET',
        url.encode('utf-8'),
        Headers({'User-Agent': ['Twisted Web Client']}),
        None)
    d.addCallback(readBody)
    return d

@inlineCallbacks
def verify_jwt(token):
    # Fetch the JWKS
    jwks_response = yield fetch_jwks(jwks_url)
    jwks = json.loads(jwks_response)

    # Decode the token header to get the key ID
    headers = jwt.get_unverified_headers(token)
    kid = headers['kid']

    # Find the key in the JWKS
    key = None
    for jwk_key in jwks['keys']:
        if jwk_key['kid'] == kid:
            key = jwk_key
            break

    if key is None:
        raise Exception("Public key not found in JWKS")

    # Construct a key for verification
    public_key = jwk.construct(key)

    # Verify the token
    message, encoded_signature = token.rsplit('.', 1)
    decoded_signature = base64url_decode(encoded_signature.encode('utf-8'))

    if not public_key.verify(message.encode('utf-8'), decoded_signature):
        raise jwt.JWTError("Signature verification failed")

    # Decode the token to get the payload
    payload = jwt.decode(token, key, algorithms=['RS256'], audience='<your-expected-audience>', issuer='https://<keycloak-domain>/auth/realms/<realm-name>')

    returnValue(payload)

@inlineCallbacks
def main():
    try:
        payload = yield verify_jwt(token)
        print("Token is valid:", payload)
    except Exception as e:
        print("Failed to verify token:", e)
    finally:
        reactor.stop()

if __name__ == "__main__":
    reactor.callWhenRunning(main)
    reactor.run()