#!/usr/bin/python """ (C) Copyright European Space Agency, 2026 Owner: European Space Agency Author: K. Panitzek Version: 2.1 Last updated: 31/03/2026 Change log: Version | Date | Reason for change 1.0 | 04/03/2021 | Initial version 1.1 | 08/10/2021 | Change of cookie name 1.2 | 28/03/2022 | Improve consent handling as the consent is now stored in OpenAM 1.3 | 20/10/2022 | Adjust to correct HAPI capabilities response 2.0 | 25/02/2026 | Added support for Keycloak authentication via OIDC access token 2.1 | 31/03/2026 | Remove authentication via OpenAM and consolidation Prerequisites: - Python 3 and Python libraries 'requests' and 'json' installed on the system - OIDC Client-ID and -Secret must have been created and configured Purpose: This Python script/module will attempt to retrieve an OIDC access token for accessing the SWE HAPI server hosted at ESA's SWE Portal. If a Client-ID and a Client-Secret (and a scope) have been provided, these credentials will be used to obtain an access token from the SSO server (Keycloak). The variables Client-ID and Client-Secret refer to the OpenId client's credentials when authenticating against Keycloak. The Scope determines the extent of the requested access permissions. It is a list of white space separated string literals. It must contain at least the Client-ID of the protected resources the token is being requested for (e.g. 'swe_hapiserver' for the SWE HAPI Server). Keycloak will embed this information into the token, to prevent potential misuse. While this script defaults to obtain an access token for the SWE HAPI Server, it can also be used to request an access token for the SWE Content Proxy by setting the scope to 'swe_contentproxy' (option -y). More details can be found on the SWE HAPI Server page on ESA's SWE Portal at: https://swe.ssa.esa.int/swe-hapi-server The script is also intended to be used as a template and adapted to work with other services than HAPI. It provides the following methods: - get_oauth2_access_token(client_id, client_secret): Retrieves an oauth2 access token using client id, client secret. - is_token_valid(token_incl_metadata, response_timestamp): Checks if the retrieved access token is still valid given the timestamp the token was received. - extract_access_token(token_incl_metadata): Extracts the access token from the returned response (access token incl all metadata). - get_hapi_capabilities(access_token): Calls HAPI capabilities endpoint using the provided token for authentication. The main method can be executed to test the entire script. To do so, follow the usage instructions below. Usage: get_hapi_access_token.py [-h] -i CLIENTID -x CLIENTSECRET [-y SCOPE] [-c PORTALURL] [-s] [-t] [-v] This script will attempt to retrieve an OIDC access token for accessing the SWE HAPI server hosted at ESA's SWE Portal using the provided Client-ID and Client-Secret. Options: -h, --help Show this help message and exit -i CLIENTID, --clientid CLIENTID Set the Client-ID to access HAPI -x CLIENTSECRET, --clientsecret CLIENTSECRET Set the Client-Secret to access HAPI. -y SCOPE, --scope SCOPE Set the scope, a space-separated list, of the protected resource to access, e.g., 'openid swe_hapiserver'. This will override the default value: 'swe_hapiserver' -c PORTALURL, --portalurl PORTALURL Set a custom SWE Portal URL where HAPI is hosted. This will override the default value: 'https://swe.ssa.esa.int' -s, --silent Disable printing of error messages. -t, --test If provided, the script will access the HAPI /capabilities endpoint to test the obtained access token. -v, --verbose Enable verbose mode. """ import argparse import datetime from datetime import datetime, timezone from http import HTTPStatus import json import requests # Global variables to control authentication and authorisation for the respective environments PORTAL_URL="https://swe.ssa.esa.int" KEYCLOAK_URL="https://sso.s2p.esa.int/realms/swe/protocol/openid-connect/token" SCOPE='swe_hapiserver' def get_oauth2_access_token(client_id: str, client_secret: str, scopes = SCOPE, auth_url = KEYCLOAK_URL, silent=True): """ Retrieves an oauth2 access token using client id, client secret and scopes. :param client_id: Client id to use for authentication :param client_secret: Client secret to use for authentication :param scopes: Scopes to request, defaults to 'swe_hapiserver' :param auth_url: URL to authentication server. Defaults to 'https://sso.s2p.esa.int/realms/swe/protocol/openid-connect/token' :param silent: Print error messages if False, defaults to True :return: The access token incl all metadata, the timestamp at which the token was retrieved """ try: response = requests.post(auth_url, data={ "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, "scope": scopes }) if response.status_code != HTTPStatus.OK: if not silent: print(response) return None, None return json.loads(response.content), datetime.now(timezone.utc) except Exception as e: if not silent: print("Requesting access token has failed: " + str(e)) return None, None def is_token_valid(token_incl_metadata, response_timestamp): """ Checks if the retrieved access token is still valid given the timestamp the token was received :param token_incl_metadata: The obtained access token including all metadata :param response_timestamp: The timestamp at which this access token was retrieved :return: True, if the token is still valid """ return (datetime.now(timezone.utc) - response_timestamp).seconds < token_incl_metadata.get("expires_in") def extract_access_token(token_incl_metadata): """ Extracts the access token from the returned response (access token incl all metadata) :param token_incl_metadata: The obtained access token including all metadata :return: The access token that can be used to access HAPI endpoints """ return token_incl_metadata.get('access_token') def get_hapi_capabilities(access_token: str, silent=True): """ Calls the HAPI /capabilities endpoint using the provided token for authentication :param access_token: The access token without the metadata :param silent: Print error messages if False :return: True and the parsed HAPI capabilities response, else False """ try: # send request and supply access token via bearer token auth test_response = requests.get(PORTAL_URL + "/hapi/capabilities", headers={ "Authorization": f"Bearer {access_token}" }) if test_response.status_code != HTTPStatus.OK: if not silent: print(test_response) return False, {} return _parse_hapi_response(test_response) except Exception as exc: if not silent: print(exc) return False, {} def _parse_hapi_response(response, silent=True): """ Helper function to parse and validate the provided response to the HAPI /capabilities endpoint :param response: The response returned from the request to the HAPI /capabilities endpoint :param silent: Print error messages if False :return: True of the validation was successful and the returned capabilities, else False. """ try: # extract the capabilities from the response capabilities = json.loads(response.content) hapi_version = capabilities["HAPI"] status = capabilities["status"] # if the capabilities are as expected, return True along with the capabilities dict if hapi_version != "" and status != {} and status["message"] == "OK": return True, capabilities except Exception as exc: if not silent: print(exc) return False, {} def main(): parser = argparse.ArgumentParser(description="This script will attempt to retrieve an OIDC access token for accessing the SWE HAPI server hosted at ESA's SWE Portal using the provided Client-ID and Client-Secret.") parser.add_argument("-i", "--clientid", required=True, help="Set the Client-ID to access HAPI") parser.add_argument("-x", "--clientsecret", required=True, help="Set the Client-Secret to access HAPI.") parser.add_argument("-y", "--scope", required=False, help=f"Set the scope, a space-separated list, of the protected resource to access, e.g., 'openid swe_hapiserver'. This will override the default value: '{SCOPE}'") parser.add_argument("-c", "--portalurl", required=False, help=f"Set a custom SWE Portal URL where HAPI is hosted. This will override the default value: '{PORTAL_URL}'") parser.add_argument("-s", "--silent", action="store_true", required=False, help="Disable printing of error messages.") parser.add_argument("-t", "--test", action="store_true", required=False, help="If provided, the script will access the HAPI /capabilities endpoint to test the obtained access token.") parser.add_argument("-v", "--verbose", action="store_true", required=False, help="Enable verbose mode.") args = parser.parse_args() client_id = args.clientid client_secret = args.clientsecret scope = SCOPE if args.scope: scope = args.scope silent = args.silent test = args.test verbose = args.verbose if verbose: print(f"Retrieving access token for scope '{SCOPE}'") token_incl_metadata, response_timestamp = get_oauth2_access_token(client_id, client_secret, scope, silent=silent) if token_incl_metadata is not None: if verbose: print(f"Authentication successful. Received access token for scope '{SCOPE}':") print(json.dumps(token_incl_metadata, indent=4)) print() if is_token_valid(token_incl_metadata, response_timestamp): access_token = extract_access_token(token_incl_metadata) if test: if verbose: print("Testing access token on HAPI /capabilities endpoint:") has_succeeded, capabilities = get_hapi_capabilities(access_token, silent=silent) if has_succeeded: print(capabilities) else: print("Capabilities test failed. The access token might have expired in the meantime.") print() if verbose: print("Access Token to be used for subsequent HAPI calls:") print(access_token) else: print("Token has expired and needs to be renewed") else: print("Authentication failed") if __name__ == "__main__": main()