Merge pull request #3 from itk-dev-rpa/release/1.0.0

Release/1.0.0

Author: ghbm-itk

.github/workflows/Linting.yml

@@ -4,7 +4,7 @@ on: [pull_request]
 
 jobs:
   build:
-    runs-on: windows-latest
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ["3.11"]

README.md

@@ -1,3 +1,49 @@
-# python-serviceplatformen
+# python_serviceplatformen
 
-Python modules to easier use Kombit's Serviceplatformen API
+This project is made to hopefully make it easier to use Kombit's Serviceplatformen API.
+
+## Certificates and authentication
+
+To use Kombit's Serviceplatformen you need a valid OCES3 certificate registered for your project.
+Ask your local Kombit systems architect for help with this.
+
+You need specific access to each service on Serviceplatformen you want to use.
+One certificate can have access to multiple services.
+
+This project needs your certificate to be in a unified PEM format. That is
+with the public and private key in a single file.
+
+If your certificate is in P12-format you can convert it using openssl:
+
+```bash
+openssl pkcs12 -in Certificate.p12 -out Certificate.pem -nodes
+```
+
+Due to limitations in Python's implementation of SSL your certificate needs to exist as
+a file on the system while using this library.
+
+When your certificate is registered and in PEM-format, you simply hand it to the KombitAccess
+class and it will handle the rest for you.
+
+```python
+from python_serviceplatformen.authentication import KombitAccess
+ka = KombitAccess(cvr=cvr, cert_path=cert_path)
+```
+
+## Tests
+
+This project contains automated tests in the "tests" folder.
+To run these test you need to install the developer dependecies:
+
+```bash
+pip install python_serviceplatformen[dev]
+```
+
+### Environment variables
+
+Create a .env file in the project folder and fill out these variables:
+
+```yaml
+KOMBIT_TEST_CVR = "XXXXXXXX"  # The cvr of the organization who owns the certificate
+KOMBIT_TEST_CERT_PATH = "C:\something\Certificate.pem"  # The path to the certificate file
+```

changelog.md

@@ -6,3 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+
+## [1.0.0] - 19-09-2024
+
+### Added
+
+- Module for authenticating towards Kombit's Serviceplatform API.
+- Function for checking if someone is registered for Digital Post or NemSMS.
+
+[Unreleased]: https://github.com/itk-dev-rpa/python-serviceplatformen/compare/1.0.0...HEAD
+[1.0.0]: https://github.com/itk-dev-rpa/python-serviceplatformen/releases/tag/1.0.0

pyproject.toml

@@ -3,8 +3,8 @@ requires = ["setuptools>=65.0"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "python-serviceplatformen"
-version = "0.0.1"
+name = "python_serviceplatformen"
+version = "1.0.0"
 authors = [
   { name="ITK Development", email="[email protected]" },
 ]
@@ -16,12 +16,13 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
 ]
 dependencies = [
-  "requests == 2.*"
+  "requests == 2.*",
+  "cryptography"
 ]
 
 [project.urls]
-"Homepage" = "https://github.com/itk-dev-rpa/python-serviceplatformen"
-"Bug Tracker" = "https://github.com/itk-dev-rpa/python-serviceplatformen/issues"
+"Homepage" = "https://github.com/itk-dev-rpa/python_serviceplatformen"
+"Bug Tracker" = "https://github.com/itk-dev-rpa/python_serviceplatformen/issues"
 
 [project.optional-dependencies]
 dev = [

python-serviceplatformen/__init__.py renamed to python_serviceplatformen/__init__.py

@@ -0,0 +1,159 @@
+"""This module contains the KombitAccess class used to authenticate against the
+Kombit API.
+"""
+
+from datetime import datetime, timedelta
+
+import requests
+from requests.exceptions import HTTPError
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+
+
+# pylint: disable-next=too-few-public-methods
+class KombitAccess:
+    """An object that handles access to the Kombit api."""
+    cvr: str
+    cert_path: str
+    _access_tokens: dict[str, (str, datetime)]
+    test: bool
+    environment: str
+
+    def __init__(self, cvr: str, cert_path: str, test: bool = False) -> None:
+        """Create a new Kombit Access object.
+
+        Args:
+            cvr: The cvr number of the organisation making the calls.
+            cert_path: The path to the certificate in unified pem-format.
+            test: Whether to use the test environment or not.
+        """
+        self.cvr = cvr
+        self.cert_path = cert_path
+        self._access_tokens = {}
+        self.test = test
+
+        if test:
+            self.environment = "https://exttest.serviceplatformen.dk"
+        else:
+            self.environment = "https://prod.serviceplatformen.dk"
+
+    def get_access_token(self, entity_id: str) -> str:
+        """Get an access token to the api endpoint with the given entity id.
+        If an access token already exists for the endpoint it is reused.
+
+        Args:
+            entity_id: The entity id of the endpoint.
+            test: Whether to use the test api or not.
+
+        Returns:
+            An access token to be used in api calls.
+
+        Raises:
+            HTTPError: If an access token couldn't be obtained for the given entity id.
+        """
+        if entity_id in self._access_tokens and datetime.now() < self._access_tokens[entity_id][1]:
+            return self._access_tokens[entity_id][0]
+
+        try:
+            saml_token = _get_saml_token(self.cvr, self.cert_path, entity_id, test=self.test)
+            access_token = _get_access_token(saml_token, self.cert_path, test=self.test)
+            self._access_tokens[entity_id] = access_token
+            return access_token[0]
+        except HTTPError as exc:
+            raise ValueError(f"Couldn't obtain access token for {entity_id}") from exc
+
+
+def _get_saml_token(cvr: str, cert_path: str, entity_id: str, test: bool) -> str:
+    """Get a SAML token for the endpoint with the given entity id.
+
+    Args:
+        cvr: The cvr number of the organisation making the calls.
+        cert_path: The path to the certificate in unified pem-format.
+        entity_id: The entity id of the endpoint.
+        test: Whether to use the test api or not.
+
+    Returns:
+        A SAML token as a string.
+    """
+    use_key = _extract_first_certificate(cert_path)
+
+    if test:
+        url = "https://adgangsstyring.eksterntest-stoettesystemerne.dk/runtime/api/rest/wstrust/v1/issue"
+    else:
+        url = "https://adgangsstyring.stoettesystemerne.dk/runtime/api/rest/wstrust/v1/issue"
+
+    payload = {
+        "TokenType": "http://docs.oasis-open.org/wss/oasis-wss-saml-token-profile-1.1#SAMLV2.0",
+        "RequestType": "http://docs.oasis-open.org/ws-sx/ws-trust/200512/Issue",
+        "KeyType": "http://docs.oasis-open.org/ws-sx/ws-trust/200512/PublicKey",
+        "AnvenderKontekst": {
+            "Cvr": cvr
+        },
+        "UseKey": use_key,
+        "AppliesTo": {
+            "EndpointReference": {
+                "Address": entity_id
+            }
+        },
+        "OnBehalfOf": None
+    }
+
+    response = requests.post(url, json=payload, cert=cert_path, timeout=10)
+    response.raise_for_status()
+
+    return response.json()['RequestedSecurityToken']['Assertion']
+
+
+def _get_access_token(saml_token: str, cert_path: str, test: bool) -> tuple[str, datetime]:
+    """Get an access token for the given SAML context.
+
+    Args:
+        saml_token: The SAML token to get the access token for.
+        cert_path: The path to the certificate in unified pem-format.
+        test: Whether to use the test api or not.
+
+    Returns:
+        The access token as a string and the expiry datetime of the token.
+    """
+    if test:
+        url = "https://exttest.serviceplatformen.dk/service/AccessTokenService_1/token"
+    else:
+        url = "https://prod.serviceplatformen.dk/service/AccessTokenService_1/token"
+
+    payload = {
+        "saml-token": saml_token
+    }
+
+    response = requests.post(url, data=payload, cert=cert_path, timeout=10)
+    response.raise_for_status()
+    response = response.json()
+
+    access_token = f"{response['token_type']} {response['access_token']}"
+    expiry_time = datetime.now() + timedelta(seconds=response['expires_in']) - timedelta(minutes=5)
+
+    return access_token, expiry_time
+
+
+def _extract_first_certificate(pem_file: str) -> str:
+    """Extract the first certificate from a certificate file.
+
+    Args:
+        pem_file: The path of the pem certificate.
+
+    Raises:
+        ValueError: If the certificate couldn't be parsed.
+
+    Returns:
+        The first certificate in the file as a single line string.
+    """
+    with open(pem_file, "rb") as f:
+        pem_data = f.read()
+
+    try:
+        cert = x509.load_pem_x509_certificate(pem_data, default_backend())
+        first_cert = cert.public_bytes(encoding=serialization.Encoding.PEM).decode("utf-8")
+        first_cert_single_line = "".join(first_cert.splitlines()[1:-1])
+        return first_cert_single_line
+    except Exception as e:
+        raise ValueError("Error parsing certificate") from e

python_serviceplatformen/authentication.py

@@ -0,0 +1,10 @@
+"""This module contains function to help with datetimes in the Kombit API."""
+
+from datetime import datetime
+
+
+def format_datetime(_datetime: datetime) -> str:
+    """Convert a datetime object to a string with the format:
+    %Y-%m-%dT%H:%M:%SZ
+    """
+    return _datetime.strftime('%Y-%m-%dT%H:%M:%SZ')

python_serviceplatformen/date_helper.py

@@ -0,0 +1,43 @@
+"""This moduel contains helper functions to use the SF1601 Kombit API.
+https://digitaliseringskataloget.dk/integration/sf1601
+"""
+
+import urllib.parse
+import uuid
+from datetime import datetime
+from typing import Literal
+
+import requests
+
+from python_serviceplatformen.authentication import KombitAccess
+from python_serviceplatformen.date_helper import format_datetime
+
+
+def is_registered(cpr: str, service: Literal['digitalpost', 'nemsms'], kombit_access: KombitAccess) -> bool:
+    """Check if the person with the given cpr number is registered for
+    either Digital Post or NemSMS.
+
+    Args:
+        cpr: The cpr number of the person to look up.
+        service: The service to look up for.
+        kombit_access: The KombitAccess object used to authenticate.
+
+    Returns:
+        True if the person is registered for the selected service.
+    """
+    url = urllib.parse.urljoin(kombit_access.environment, "service/PostForespoerg_1/")
+    url = urllib.parse.urljoin(url, service)
+
+    parameters = {
+        "cprNumber": cpr
+    }
+
+    headers = {
+        "X-TransaktionsId": str(uuid.uuid4()),
+        "X-TransaktionsTid": format_datetime(datetime.now()),
+        "authorization": kombit_access.get_access_token("http://entityid.kombit.dk/service/postforespoerg/1")
+    }
+
+    response = requests.get(url, params=parameters, headers=headers, timeout=10)
+    response.raise_for_status()
+    return response.json()['result']

python_serviceplatformen/digital_post.py

@@ -0,0 +1,39 @@
+"""Tests of Kombit API authentication."""
+import unittest
+import os
+
+from dotenv import load_dotenv
+
+from python_serviceplatformen.authentication import KombitAccess
+
+load_dotenv()
+
+# We don't care about duplicate code in tests
+# pylint: disable=R0801
+
+
+class KombitAuthTest(unittest.TestCase):
+    """Test authentication against the Kombit API."""
+
+    def test_kombit_access(self):
+        """Test authentication."""
+        cvr = os.environ["KOMBIT_TEST_CVR"]
+        cert_path = os.environ["KOMBIT_TEST_CERT_PATH"]
+        ka = KombitAccess(cvr=cvr, cert_path=cert_path, test=True)
+
+        # Test getting a token
+        token = ka.get_access_token("http://entityid.kombit.dk/service/postforespoerg/1")
+        self.assertIsInstance(token, str)
+        self.assertGreater(len(token), 0)
+
+        # Test reuse of token
+        token2 = ka.get_access_token("http://entityid.kombit.dk/service/postforespoerg/1")
+        self.assertEqual(token, token2)
+
+        # Test getting a nonsense token
+        with self.assertRaises(ValueError):
+            ka.get_access_token("FooBar")
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_auth.py

@@ -0,0 +1,41 @@
+"""Tests of the Kombit Digital Post API."""
+import unittest
+import os
+
+from dotenv import load_dotenv
+
+from python_serviceplatformen.authentication import KombitAccess
+from python_serviceplatformen import digital_post
+
+load_dotenv()
+
+# We don't care about duplicate code in tests
+# pylint: disable=R0801
+
+
+class DigitalPostTest(unittest.TestCase):
+    """Test Digital Post functionality in the Kombit API."""
+
+    def test_is_registered(self):
+        """Test authentication."""
+        cvr = os.environ["KOMBIT_TEST_CVR"]
+        cert_path = os.environ["KOMBIT_TEST_CERT_PATH"]
+        ka = KombitAccess(cvr=cvr, cert_path=cert_path, test=True)
+
+        # Fictional test cpr
+        cpr = "2611740000"
+
+        result = digital_post.is_registered(cpr=cpr, service="digitalpost", kombit_access=ka)
+        self.assertTrue(result)
+
+        result = digital_post.is_registered(cpr=cpr, service="nemsms", kombit_access=ka)
+        self.assertFalse(result)
+
+        # Test with nonsense.
+        # This should result in False
+        result = digital_post.is_registered(cpr="FooBar", service="digitalpost", kombit_access=ka)
+        self.assertFalse(result)
+
+
+if __name__ == '__main__':
+    unittest.main()