Python API : ib1.openenergy.support

Python docs and source for this package

External APIs

class ib1.openenergy.support.AccessTokenValidator(client_id: str, private_key: str, certificate: str, issuer_url: str = 'https://matls-auth.directory.energydata.org.uk/', client_cert_parser=None)

Perform checks on a presented bearer token as defined in section 6.2.1 here https://openid.net/specs/openid-financial-api-part-1-1_0-final.html#accessing-protected-resources

Uses https://tools.ietf.org/html/rfc7662 - OAuth 2.0 Token Introspection to check a supplied bearer token against an introspection endpoint for part 6.2.1.13

__init__(client_id: str, private_key: str, certificate: str, issuer_url: str = 'https://matls-auth.directory.energydata.org.uk/', client_cert_parser=None)

Create a new access token validator. In this context the data provider attempting to validate an access token is acting as a client to the directory’s API, so client_id, private_key and certificate are those of the data provider and not the client requesting data from it. These are not the transport keys, they’re the ones issued by the directory.

Parameters

• client_id – OAuth client ID of the data provider, used to authenticate with the introspection endpoint

• private_key – Location of the private key of the data provider, used in the client auth for the introspection endpoint

• certificate – Location of the public key of the data provider, used in the client auth for the introspection endpoint

• issuer_url – URL of an authorization server, i.e. for the raidiam UAT sandbox this is https://matls-auth.directory.energydata.org.uk/ - uses https://openid.net/specs/openid-connect-discovery-1_0.html part 4 to discover the token introspection endpoint

• client_cert_parser – A zero argument function which returns an X509 object for the active client certificate, or none if no certificate is present. Defaults to a simple implementation that pulls the cert out of the flask environment as provided by the local dev mode runner in flask_ssl_dev.py, but should be replaced when running in production mode behind e.g. nginx

inspect_token(token: str) → Dict

Send an access token to the introspection endpoint, returning the introspection response

Parameters

token – access token as received as authorization in a request to the data provider’s API

Returns

object containing parsed JSON response from the introspection endpoint

introspects(f=None, scope=None)

Build a decorator that can be used on flask routes to automatically introspect on any provided bearer tokens, passing the resulting object into g.introspection_response. If the introspection indicates a failed validation, the underlying route will not be called at all and an appropriate error response will be sent.

Introspection fails if:

1. Querying the token introspection endpoint fails

2. A token is returned with active: false

3. Scope is specified, and the required scope is not in the token scopes

4. Issued time is in the future

5. Expiry time is in the past

6. Certificate binding is enabled (default) and the fingerprint of the presented client cert isn’t a match for the claim in the introspection response

If introspection succeeds, the decorated function is called and the Date and x-fapi-interaction-id headers injected into the response before returning.

class ib1.openenergy.support.FAPISession(client_id, issuer_url, requested_scopes, private_key, certificate, jwt_bearer_email=None, signing_private_key=None, retries=3)

Similar to a requests session, but handles management of a single access token. It acquires the token when required, manages token expiry etc. Implicitly uses client-credentials, there’s no user consent or similar involved.

Can also be used as a requests authenticator

__init__(client_id, issuer_url, requested_scopes, private_key, certificate, jwt_bearer_email=None, signing_private_key=None, retries=3)

Build a new FAPI session. This doesn’t immediately trigger any requests to the token endpoint, these are made when the session is accessed, and only if needed.

Parameters

• issuer_url – URL of an authorization server, i.e. for the raidiam UAT sandbox this is https://matls-auth.directory.energydata.org.uk/

• requested_scopes – Scopes requested for this session

• private_key – Location of private key used for MTLS, and to sign

• certificate – Location of certificate used for MTLS

• jwt_bearer_email – Defaults to None. If specified, this should be the email address of a user to impersonate. This will then switch the client to jwt-bearer grant type. This will only work if the corresponding client has been provisioned appropriately in the directory itself, otherwise this will fail. It should only ever be used by our internal Open Energy clients needing to write to the directory, and can be entirely ignored by other users.

• signing_private_key – Default to None, must be provided if jwt_bearer_email is set. Path to private key set as a signing key in the directory.

• retries – Number of retries that will be used when accessing GET endpoints through the underlying plain and FAPI enabled sessions within this object. Defaults to 3, set to 0 to disable retries. Requests which respond with statii in [429, 502, 503, 504] will be retried, exponential back-off is applied to avoid overwhelming resources.

clear_token()

Explicitly clear the current token, if present. This will force a refresh the next time the session is accessed.

property introspection_response: Dict

Introspect on our own token, useful to see what the resource server will see when it asks about this client.

property session

A requests session configured to automatically acquire tokens when needed and to use MTLS

class ib1.openenergy.support.OpenIDConfiguration(token_endpoint: str, introspection_endpoint: str, issuer: str, scopes_supported: List[str], grant_types_supported: List[str])

Information returned from the .well-known/openid-configuration document by an OpenID provider such as the raidiam authz service.

__init__(token_endpoint: str, introspection_endpoint: str, issuer: str, scopes_supported: List[str], grant_types_supported: List[str]) → None

Initialize self. See help(type(self)) for accurate signature.

static oidc_configuration_url(issuer_url: str)

Implements the logic in 4.1 of https://openid.net/specs/openid-connect-discovery-1_0.html to find the URL for the configuration document which can populate an instance of OpenIDConfiguration

Parameters

issuer_url – Base URL of the issuer

Returns

URL of the configuration document

class ib1.openenergy.support.RaidiamDirectory(fapi: ib1.openenergy.support.FAPISession, base_url: str = 'https://matls-dirapi.directory.energydata.org.uk/')

Encapsulates access to the Raidiam Directory, currently just the read API. Parses responses and builds the appropriate dataclasses from the ib1.openenergy.support.raidiam module.

__init__(fapi: ib1.openenergy.support.FAPISession, base_url: str = 'https://matls-dirapi.directory.energydata.org.uk/')

Initialize self. See help(type(self)) for accurate signature.

organisations() → List[ib1.openenergy.support.raidiam.Organisation]

Get all Organisation entities within the directory

ib1.openenergy.support.build(d: Dict, cls: Type[ib1.openenergy.support.D], date_format_string='%Y-%m-%dT%H:%M:%S.%fZ') → ib1.openenergy.support.D

Build a dataclass from a dict, massaging CamelCase form into the more normal pythonic_representation, then filtering by properties available to the dataclass constructor before using the filtered set to create a new instance of the dataclass. Handles entries which are typed to other dataclasses recursively including List[OtherDataClass] types.

Parameters

• d – Dict containing properties to pass to constructor

• cls – Class, typically a dataclass, to receive properties. Should be the class of the root object.

• data_format_string – Format string to use when parsing dates into datetime objects, defaults to the one we’re seeing in the Raidiam directory, i.e. “%Y-%m-%dT%H:%M:%S.%fZ”

Returns

Instance of cls configured from the supplied dict

ib1.openenergy.support.build_error_response(error=None, code=400, scope=None, description=None, uri=None)

RFC 6750 has a slightly odd way to complain about invalid tokens! This is used in the token authenticator class to build responses for invalid or missing tokens.

Parameters

• error – One of ‘invalid request’, ‘invalid token’, or ‘insufficient scope’

• code – HTTP status code

• description – Description of the error

• uri – URL of the page describing the error in detail

• scope – Scopes required to access the protected resource

ib1.openenergy.support.httpclient_logging_patch(level=10)

Enable HTTPConnection debug logging to the logging framework

ib1.openenergy.support.nginx_cert_parser()

A certificate parser to be used with the AccessTokenValidator when running behind nginx or another similar proxy which terminates SSL connections and can be configured to push the presented client certificate into a header. In this case we use the header X-OE-CLIENT-CERT, this function removes any errant tab characters (introduced by nginx for some reason) and parses the contents of this header as a PEM format certificate.

Returns

A parsed x509 Certificate object, or None if no cert presented in the header

Metadata APIs

API to handle the metadata format described in Data Set Metadata

ib1.openenergy.support.metadata.DC = 'http://purl.org/dc/terms/'

Dublin Core namespace

ib1.openenergy.support.metadata.DCAT = 'http://www.w3.org/ns/dcat#'

Data Catalogue namespace

class ib1.openenergy.support.metadata.JSONLDContainer(d: Dict)

Wraps up the data structure returned by jsonld.expand and adds some convenience methods to query properties within it

__init__(d: Dict)

Initialize self. See help(type(self)) for accurate signature.

get(namespace: str, term: str, default=None)

Get a property, handles looking for the @value entries within an expanded JSON-LD dictionary

Parameters

• namespace – namespace for term to find

• term – term within that namespace

• default – default value to return if term isn’t present, defaults to None

Returns

value of term, can be single item if only one value present or list if multiple

require_values(d: Dict[str, List[str]])

Require that this container has the specified values, defined as a dict of namespace to list of terms.

Parameters

d – Dict of str namespace to list of str terms that must be present

Raises

ValueError if any specified values are not present in this container

property type

@type of the entity described

class ib1.openenergy.support.metadata.Metadata(d: Dict)

Representation of the information held in a data set metadata file, as defined in https://icebreakerone.github.io/open-energy-technical-docs/main/metadata.html - this implementation will track the spec, but may not be complete, we’ll build capabilities into it as and when we need them.

Currently models the content part of the metadata file as a JSONLDContainer

__init__(d: Dict)

Create a new metadata container. Currently just parses the content section of the metadata file.

Parameters

d – a dict containing the four top level keys from the metadata spec

Raises

ValueError – if the structure of the dict is invalid in some way

property data_sensitivity_class: str

content / oe:sensitivityClass [OE-O|OE-SA|OE-SB]

property description

content / dc:description

property keywords: List[str]

content / dcat:keywords

property stable_identifier: str

content / oe:dataSetStableIdentifier

property title: str

content / dc:title

property version

content / dcat:version

property version_notes

content / dcat:versionNotes

class ib1.openenergy.support.metadata.MetadataLoadResult(location: Optional[str] = None, error: Optional[str] = None, exception: Optional[Exception] = None, metadata: List[ib1.openenergy.support.metadata.Metadata] = <factory>, server: Optional[ib1.openenergy.support.raidiam.AuthorisationServer] = None)

Information about the process of loading a metadata file from a URL or file location along with the results. This is used instead of raising exceptions during the load process in order to provide better reporting with mappings between org IDs and problems with their respective metadata files.

__init__(location: Optional[str] = None, error: Optional[str] = None, exception: Optional[Exception] = None, metadata: List[ib1.openenergy.support.metadata.Metadata] = <factory>, server: Optional[ib1.openenergy.support.raidiam.AuthorisationServer] = None) → None

Initialize self. See help(type(self)) for accurate signature.

ib1.openenergy.support.metadata.OE = 'http://energydata.org.uk/oe/terms/'

Open Energy ontology namespace

ib1.openenergy.support.metadata.load_metadata(server: Optional[ib1.openenergy.support.raidiam.AuthorisationServer] = None, url: Optional[str] = None, file: Optional[str] = None, session=None, convert_tabs_to_spaces=True, **kwargs) → ib1.openenergy.support.metadata.MetadataLoadResult

Load metadata from a URL.

Parameters

• server – AuthorisationServer from which this url was retrieved, or None if fetching directly

• url – url from which to load metadata

• file – file path from which to load metadata, use either this or url, not both

• session – if specified, use this requests.Session, if not, create a new one

• convert_tabs_to_spaces – normally YAML is invalid if it uses tab characters as indentation. If this argument is set to true, a second attempt will be made to parse the file if a scanner error occurs, first doing a global search and replace to change all tab characters to double spaces. Defaults to False, as this isn’t really ‘allowed’ according to the spec.

• kwargs – any additional arguments to pass into the get request

Returns

a MetadataLoadResult containing a report on the process, including actual Metadata objects if the load succeeded and found any metadata.

ib1.openenergy.support.metadata.load_yaml_from_bytes(b: bytes, convert_tabs_to_spaces=True)

Attempt to load YAML from a set of bytes.

Parameters

• b – bytes to load

• convert_tabs_to_spaces – if True (default), an initial failure to load YAML will be retried after converting all tab characters in the input to double spaces. This is done after converting the bytes to a string with UTF8 encoding, and after the tabs are stripped the string is encoded back to UTF8 bytes before passing back to the yaml loader

Returns

yaml parsed as a dict

Raises

YAMLError if unable to parse the input bytes

Gunicorn support APIs

Support for running data providers within the Gunicorn WSGI container

Support for gunicorn with client certificates, with much help from a blog at https://eugene.kovalev.systems/blog/flask_client_auth

ib1.openenergy.support.gunicorn.CERT_NAME = 'X-OE-CLIENT-CERT'

Name for the header containing the client certificate as a BASE64 encoded DER file

class ib1.openenergy.support.gunicorn.ClientAuthApplication(app, cert_path, key_path, hostname='localhost', port='443', num_workers=4, timeout=30)

GUnicorn application using the custom SSL worker. Uses certifi for its CA store. This is a helper class, mostly useful when you need to run a data provider as part of a unit test, all it really does is remove the need for a gunicorn.conf.py configuration file. See this blog for more details on how to run a data provider within a test context using this class.

__init__(app, cert_path, key_path, hostname='localhost', port='443', num_workers=4, timeout=30)

Create a new application runner

Parameters

• app – WSGP app to run

• cert_path – Path to the server certificate

• key_path – Path to the server private key

• hostname – Hostname, defaults to ‘localhost’

• port – Port, defaults to 443

• num_workers – Number of concurrent workers, defaults to 4

• timeout – Timeout, defaults to 30

load_config()

Overrides default configuration with that defined in self.options

class ib1.openenergy.support.gunicorn.CustomSyncWorker(age, ppid, sockets, app, timeout, cfg, log)

Push x509 certificate from SSL context into the named header in BASE64 encoded format. Uses the header name defined as CERT_NAME

ib1.openenergy.support.gunicorn.LOG = <Logger ib1.openenergy.support.gunicorn (WARNING)>

Log to ib1.openenergy.support.gunicorn

ib1.openenergy.support.gunicorn.gunicorn_cert_parser() → cryptography.x509.base.Certificate

Pull x509 client cert out of the header used by the sync worker defined in this package. Header contains BASE64 encoded DER format certificate.

Use this as the client_cert_parser argument to AccessTokenValidator to allow it to pull certificates out of the named header.

SSL Development APIs

Warning

This is deprecated since 0.2.4, use the gunicorn support above instead.

class ib1.openenergy.support.flask_ssl_dev.SSLOptions(server_private_key: str = None, server_certificate: str = None, client_private_key: str = None, client_certificate: str = None, client_id: str = None)

__init__(server_private_key: Optional[str] = None, server_certificate: Optional[str] = None, client_private_key: Optional[str] = None, client_certificate: Optional[str] = None, client_id: Optional[str] = None) → None

Initialize self. See help(type(self)) for accurate signature.

ib1.openenergy.support.flask_ssl_dev.get_command_line_ssl_args(default_server_private_key='./a.key', default_server_certificate='./a.pem', default_client_private_key='./b.key', default_client_certificate='./b.pem', default_client_id='CLIENT_ID') → ib1.openenergy.support.flask_ssl_dev.SSLOptions

Parse command line arguments for SSL file paths and private key password (optional)

Parameters

• default_server_certificate – default location for app certificate, ‘./app.crt’. This is the certificate the web server is going to use to create an HTTPS endpoint, it is not the one from the directory!

• default_server_private_key – default location for app private key, ‘./app.key’. This is the private key of the app certificate, and not the one you used when creating the raidiam cert.

• default_client_certificate – default location for the client certificate

• default_client_private_key – default location for the client private key

• default_client_id – default OAuth2 client ID

Returns

parsed options object

Raises

ValueError – if any files are not specified, or not found

ib1.openenergy.support.flask_ssl_dev.run_app(app, *args, server_private_key, server_certificate, **kwargs)

Run the provided app object, parsing command line arguments to get the various SSL parameters needed for client cert validation. Checks for existence of SSL related file paths, then runs the supplied flask app. This shouldn’t be used in production, but is a quick way to spin up an app with HTTPS and client auth enabled. Any routes in the resultant app will have access to a client cert (if present) through request.environ[“peercert”].

In any real case we’d be running flask behind something like nginx, in which case client certificates are handled first by the proxy, and then generally pushed through as an additional header. This mechanism wouldn’t work in those cases, it really is just for development mode and any access to the client certs should allow for both possible mechanisms so that code works in both dev and production contexts.

Parameters

• app – flask app to run with client certificate support

• args – any additional positional arguments to pass to app.run

• server_private_key – location of private key file for app SSL

• server_certificate – app SSL certificate

• kwargs – any keyword arguments to pass to app.run, ssl_context and request_handler are already set up by this function

Internal APIs

Note

The classes below are primarily used internally within Open Energy to manage information in our membership directory and CKAN servers, they are unlikely to be of interest to third parties implementing Data Provider or Consumer components.

class ib1.openenergy.support.raidiam.AdminUser(status: str, user_email: str)

__init__(status: str, user_email: str) → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.ApiDiscoveryEndpoint(api_discovery_id: str, api_endpoint: str)

__init__(api_discovery_id: str, api_endpoint: str) → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.ApiResource(api_resource_id: str, api_family_type: str, api_version: str, api_discovery_endpoints: List[ib1.openenergy.support.raidiam.ApiDiscoveryEndpoint])

__init__(api_resource_id: str, api_family_type: str, api_version: str, api_discovery_endpoints: List[ib1.openenergy.support.raidiam.ApiDiscoveryEndpoint]) → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.AuthorisationServer(authorisation_server_id: str, organisation_id: str, auto_registration_supported: bool = False, customer_friendly_description: str = '', customer_friendly_logo_uri: str = '', customer_friendly_name: str = '', developer_portal_uri: str = '', terms_of_service_uri: str = '', open_i_d_discovery_document: str = '', payload_signing_cert_location_uri: str = '', parent_authorisation_server_id: str = '', api_resources: List[ib1.openenergy.support.raidiam.ApiResource] = None, notification_webhook: str = '', notification_webhook_status: str = '')

__init__(authorisation_server_id: str, organisation_id: str, auto_registration_supported: bool = False, customer_friendly_description: str = '', customer_friendly_logo_uri: str = '', customer_friendly_name: str = '', developer_portal_uri: str = '', terms_of_service_uri: str = '', open_i_d_discovery_document: str = '', payload_signing_cert_location_uri: str = '', parent_authorisation_server_id: str = '', api_resources: Optional[List[ib1.openenergy.support.raidiam.ApiResource]] = None, notification_webhook: str = '', notification_webhook_status: str = '') → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.Organisation(organisation_id: str, status: str, organisation_name: str, created_on: datetime.datetime, legal_entity_name: str, country_of_registration: str, company_register: str, registration_number: str, registered_name: str, city: str, postcode: str, country: str, requires_participant_terms_and_conditions_signing: bool, registration_id: str = '', address_line1: str = '', address_line2: str = '', parent_organisation_reference: str = '')

__init__(organisation_id: str, status: str, organisation_name: str, created_on: datetime.datetime, legal_entity_name: str, country_of_registration: str, company_register: str, registration_number: str, registered_name: str, city: str, postcode: str, country: str, requires_participant_terms_and_conditions_signing: bool, registration_id: str = '', address_line1: str = '', address_line2: str = '', parent_organisation_reference: str = '') → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.OrganisationAuthorityDomainClaim(organisation_authority_domain_claim_id: str, authorisation_domain_name: str, authority_id: str, authority_name: str, registration_id: str, status: str)

__init__(organisation_authority_domain_claim_id: str, authorisation_domain_name: str, authority_id: str, authority_name: str, registration_id: str, status: str) → None

Initialize self. See help(type(self)) for accurate signature.

class ib1.openenergy.support.raidiam.OrganisationContact(contact_id: str, organisation_id: str, contact_type: dict, first_name: str, last_name: str, department: str, email_address: str, phone_number: str, address_line1: str, address_line2: str, city: str, postcode: str, country: str, additional_information: str, pgp_public_key: str)

__init__(contact_id: str, organisation_id: str, contact_type: dict, first_name: str, last_name: str, department: str, email_address: str, phone_number: str, address_line1: str, address_line2: str, city: str, postcode: str, country: str, additional_information: str, pgp_public_key: str) → None

Initialize self. See help(type(self)) for accurate signature.

Support for creating and updating records in a remote CKAN instance from various organisation and metadata types from elsewhere in this library.

ib1.openenergy.support.ckan.ckan_dataset_name(org: ib1.openenergy.support.raidiam.Organisation, data_set: ib1.openenergy.support.metadata.Metadata) → str

Calculates a stable dataset ID for a given organisation and metadata object, uses the literal ‘oe’, and the ckan_legal_name forms of the organisation ID and data set stable identifier separated by ‘-‘ characters

ib1.openenergy.support.ckan.ckan_dict_from_metadata(m: ib1.openenergy.support.metadata.Metadata) → dict

Create a CKAN dict suitable for data package create or update operations from a Metadata object. Currently handles title, notes, version, tags, and adds oe:dataSensitivityClass and dcat:versionNotes to the extras dict. Tags are added without any associated vocabulary at this point.

Parameters

m – a Metadata object containing information about the data set to store or update

Returns

a dict suitable for CKAN update / create operations on data packages

ib1.openenergy.support.ckan.ckan_legal_name(s: str) → str

CKAN names must be between 2 and 100 characters long and contain only lowercase alphanumeric characters, ‘-‘ and ‘_’. This function makes a reasonable best effort to convert an arbitrary input string. Strings below 2 characters long will raise a ValueError.

Longer strings are converted to lower case, stripped of all non-alphanumeric, non underscore or dash, characters, then truncated to 100 characters if necessary.

Parameters

s – input string

Returns

valid CKAN name, as close as possible to the original

Raises

ValueError – if the string is too short

ib1.openenergy.support.ckan.update_or_create_ckan_record(org: ib1.openenergy.support.raidiam.Organisation, data_sets: List[ib1.openenergy.support.metadata.Metadata], ckan_api_key: str, ckan_url: str) → List[Dict]

Create or update records for a given datasets, each defined by a Metadata object, in the context of an Organisation from the directory. The organisation will be created if required.

Parameters

• org – Organisation to use as owner of this data set

• data_sets – list of Metadata containing information about the data sets

• ckan_api_key – api key to write to CKAN

• ckan_url – url of the CKAN instance

Returns

list of created or modified record from CKAN as dicts

Raises

NotAuthorized – if the supplied access token doesn’t have necessary permissions

Command line tools used to monitor the contents of the directory, largely for internal Icebreaker use.

ib1.openenergy.support.directory_tools.get_directory_client(parser=None) → ib1.openenergy.support.RaidiamDirectory

Parse arguments and get a directory client

Parameters

parser – Existing parser to use, None by default will create a new one

Returns

A RaidiamDirectory client