Finalize UZI PoC Q4 for YubiSign project

.editorconfig

@@ -0,0 +1,19 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+indent_style = space
+indent_size = 2
+
+[*.py]
+indent_size = 4
+max_line_length = 120
+
+[Makefile]
+indent_style = tab
+
+[*.md]
+trim_trailing_whitespace = false

.env.example

@@ -0,0 +1,4 @@
+YUBIKEY_PIN="123456"
+ACME_SERVER_DIRECTORY_URL="https://acme.proeftuin.uzi-online.irealisatie.nl/directory"
+OIDC_PROVIDER_BASE_URL="https://proeftuin.uzi-online.irealisatie.nl"
+ACME_ACCOUNT_EMAIL="[email protected]"

.github/workflows/formatlint.yml

@@ -0,0 +1,35 @@
+name: Format and lint
+
+on:
+  push:
+    branches:
+      - "main"
+  pull_request:
+    branches:
+    - "*"
+    types: [opened, synchronize, closed]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.13"]
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install developement requirements
+        run: pip install -r requirements-dev.txt
+
+      - name: Check for linting errors
+        run: ruff check . 
+
+      - name: Check for formatting errors
+        run: ruff format --check
+

.gitignore

@@ -0,0 +1,3 @@
+__pycache__
+.pytest_cache
+.env

README.md

@@ -1,45 +1,60 @@
-# Disclaimer
-This Repository is created as a PoC (Proof of Concept) as part of the project *Toekomstbestendig maken UZI*, and 
-**should not be used as is in any production environment**.
+# PoC with Yubikey
 
-# Wat doet dit?
+In order to automate certificate issuance for UZI, a PoC was done with a YubiKey and an ACME server. The keypairs are generated on the YubiKey and the certificate is issued by the ACME server. This program is designed to start with a _new_ YubiKey, meaning it should have the default PIN. This document will give you an high overview.
 
-Dit neemt een yubikey (doe maar versie 5) en maakt daarin de PIV module *leeg*
+### Steps
 
-Nadat deze leeg is worden er 4 keys aangemaakt in de yubikey
+- The YubiKey is reset: all the certificates on the device will be removed and the PIN code will be reset.
+- We will generate 4 public and private key pairs on the YubiKey. These are for PIV Authentication, Digital Signature, Key Management and Card Authentication. Next to that, the YubiKey will generate additional attestation certificates, to prove that the private key is generated on the YubiKey itself. The private keys will always remain in the YubiKey.
+- The user logs in via the chosen [authentication flow](./AUTH_FLOW.md). This returns an JWT, containing the user information.
+- Per generated key pair, an certificate signing request (CSR) is created and signed by the private key.
+- Finally, each certificate signing request with the corresponding attestation certificate is validated at the ACME server. When this is done, the server will issue an certificate for every key pair. Here, the JWT of the user is also used. This is done with the ACME server of iRealisatie. These are then saved back into the YubiKey into the corresponding slot.
 
-Er wordt contact gelegd met de rdo-acme service. En er worden 4 orders aangemaakt
+Now it is possible to use the certificate on the YubiKey to sign data.
 
-Van deze 4 orders wordt de unique-anti-replay-token meegestuurd met een uzi-labs digid login verzoek
+#### Diagram flow
 
-Er wordt een browser geopend in de applicatie zelf, daarmee log je in als zorg identiteit bij de ziekenboeg-uzi-labs
+This diagram expects that the Yubikey is already plugged in the user's computer. Next to that, it's expected that the user should use the **DigiD mock** login method.
 
-De app haalt hierna de JWT-token op bij ziekenboeg-uzi-labs waarin de 4 acmetokens zitten.
+```mermaid
+sequenceDiagram
+    actor APP
+    participant YUBIKEY
 
-Er wordt van de yubikey zelf opgehaald:
-* Het intermediate certificaat behorende bij de yubikey zoals geleverd door yubico op de yubikey zelf
+    APP->>YUBIKEY: 1. Sends request to empty the Yubikeys certificates
+    YUBIKEY-->YUBIKEY: Empties the certificates
 
-Per sleutel op de yubikey:
-* Een door de yubikey ondertekend certificaat per sleutel waarin de garantie (attestation) staat dat de sleutel echt op een yubikey is gemaakt
-* een CSR verzoek ondertekend door de aangemaakte sleutel
+    APP->>YUBIKEY: 2. Sends request to generate 4 new private key pairs
+    YUBIKEY-->YUBIKEY: 2.1 Create key pair for PIV Authentication
+    YUBIKEY-->YUBIKEY: 2.2 Create key pair for Digital Signature
+    YUBIKEY-->YUBIKEY: 2.3 Create key pair for Key Management
+    YUBIKEY-->YUBIKEY: 2.4 Create key pair for Card Authentication
 
-Per order wordt dan verstuurd:
-* De JWT
-* Het yubikey intermediate certificaat
-* Het attestation certificaat
+    create participant MAX
+    APP->>MAX: 3. Opens browser to login the user
+    MAX-->MAX: 3.1 Validates the user
+    MAX-->>APP: Returns the JWT containing the user information.
 
-De acme server controleert dan per order:
-* of het attestation certificaat van de sleutel klopt
-* het token voor order in de JWT zit
-* De JWT goed is en van een geldige uzi-cibg-labs-uitgever komt
+    APP->>APP: 4. Per generated key pair, <br> an certificate signing request (CSR)<br> is created and signed by the private key.
 
-Als dat klopt dan geeft de acme server terug dat het klopt en dan vraagt deze app in de laatste stap een certificaat aan met de eerder genoemde CSR.
+    create participant ACME_SERVER
+    loop 5. For every certificate signing request (CSR)
+        APP->>ACME_SERVER: Validate every certificate signing request with the corresponding attestation certificate
+        ACME_SERVER-->>APP: OK
+    end
 
-Als de CSR dezelfde public key heeft als in de vorige stap gecontroleerde gegevens wordt er een Labs-UZI certifcaat uitgegeven op basis van de gegevens in de JWT.
-Dit certificaat bevat de huidige UZI-Certificaten structuur.
+    loop 6. For every key pair
+        APP->>ACME_SERVER: Request certificate for every key pair, also using the users' JWT
+        ACME_SERVER-->>APP: OK
+        ACME_SERVER-->>YUBIKEY: Save certificates
+    end
 
-Als er een certificaat is opgehaald wordt dit opgeslagen op de juiste plek in de yubikey.
+```
 
-Door het laden van de yubikey pkcs11 library in de browser, office, mac os, windows of linux plekken (zoals beschreven door yubico) kan de yubikey daarna
-worden gebruikt zoals een UZIpas ook gebruikt kan worden. Voor digitaal ondertekenen van documenten, verzoeken en om in te loggen in de browser bij
-partijen die UZI certificaten login mogelijk maken.
+### Disclaimer
+
+This Repository is created as a PoC (Proof of Concept) as part of the project _Toekomstbestendig maken UZI_, and **should not be used as is in any production environment**.
+
+### Licentie
+
+This project is licensed under the [EUPL-1.2 license](./LICENSE.txt).

app/acme.py

@@ -4,18 +4,27 @@
 import requests
 from jwcrypto import jwk, jwt
 
+import urllib.parse
+
+from app.acme_directory_configuration import ACMEDirectoryConfiguration
+
 
 class Acme:
-    url = None
+    url: urllib.parse.ParseResult
     key = None
     nonce = None
     kid = None
     order = None
     certurl = None
     finalize = {}
 
-    def __init__(self, url):
-        self.url = url
+    _directory_configuration: ACMEDirectoryConfiguration
+
+    def __init__(
+        self,
+        directory_config: ACMEDirectoryConfiguration,
+    ):
+        self._directory_configuration = directory_config
 
     def debugrequest(self, protected, payload):
         print("  protected")
@@ -36,9 +45,7 @@ def gen_key(self):
         This does only generate a P-256 key for use with JWT.
         This key is only used during the session to request a certificate from ACME.
         """
-        self.key = jwk.JWK.generate(
-            kty="EC", crv="P-256", key_ops=["verify", "sign"], alg="ES256"
-        )
+        self.key = jwk.JWK.generate(kty="EC", crv="P-256", key_ops=["verify", "sign"], alg="ES256")
 
     def get_nonce(self):
         """
@@ -47,7 +54,10 @@ def get_nonce(self):
         without having to use the previous nonce. This is stored in self.nonce
         as with all updates, as every request answers with a new nonce.
         """
-        response = requests.get(self.url + "acme/new-nonce", timeout=60)
+        response = requests.get(
+            self._directory_configuration.new_nonce_url,
+            timeout=60,
+        )
         self.nonce = response.headers["Replay-Nonce"]
 
     def account_request(self, request):
@@ -58,18 +68,20 @@ def account_request(self, request):
         add the nonce (see above), url and alg and tada.wav.
         """
         print("Account Request")
+
         protected = {
             "alg": "ES256",
             "nonce": self.nonce,
-            "url": self.url + "acme/new-acct",
+            "url": self._directory_configuration.new_account_url,
             "jwk": self.key.export_public(True),
         }
         token = jwt.JWS(payload=json.dumps(request))
         token.add_signature(self.key, alg="ES256", protected=protected)
         self.debugrequest(protected, request)
         headers = {"Content-Type": "application/jose+json"}
+
         response = requests.post(
-            self.url + "acme/new-acct",
+            self._directory_configuration.new_account_url,
             data=token.serialize(),
             headers=headers,
             timeout=60,
@@ -88,18 +100,21 @@ def create_order(self, keynum, order):
         afterwards.
         """
         print("Order")
+
+        new_order_url = self._directory_configuration.new_order_url
+
         protected = {
             "alg": "ES256",
             "nonce": self.nonce,
-            "url": self.url + "acme/new-order",
+            "url": new_order_url,
             "kid": self.kid,
         }
         self.debugrequest(protected, order)
         token = jwt.JWS(payload=json.dumps(order))
         token.add_signature(self.key, alg="ES256", protected=protected)
         headers = {"Content-Type": "application/jose+json"}
         response = requests.post(
-            self.url + "acme/new-order",
+            new_order_url,
             data=token.serialize(),
             headers=headers,
             timeout=60,
@@ -132,13 +147,19 @@ def challenge(self, challengeurl):
         token = jwt.JWS(payload="")
         token.add_signature(self.key, alg="ES256", protected=protected)
         headers = {"Content-Type": "application/jose+json"}
-        response = requests.post(
-            challengeurl, data=token.serialize(), headers=headers, timeout=60
-        )
+
+        # Request the challenge per PIV-slot from the ACME-server.
+        # This will return a random token, with the status of pending.
+        #
+        # Later on, these tokens from the challenges should be contained in the users' JWT.
+        response = requests.post(challengeurl, data=token.serialize(), headers=headers, timeout=60)
+        returned_json = response.json()
+
         self.debugresponse(response)
         self.nonce = response.headers["Replay-Nonce"]
-        assert response.json()["status"] in ["pending", "valid"]
-        return response.json()["challenges"], response.json()["status"]
+
+        assert returned_json["status"] in ["pending", "valid"]
+        return returned_json["challenges"], returned_json["status"]
 
     def send_challenge_jwt(self, challenge, hw_attestation, uzi_jwt, f9_cert):
         """
@@ -167,9 +188,7 @@ def send_challenge_jwt(self, challenge, hw_attestation, uzi_jwt, f9_cert):
         }
         print("  headers")
         print(headers)
-        response = requests.post(
-            challengeurl, data=token.serialize(), headers=headers, timeout=60
-        )
+        response = requests.post(challengeurl, data=token.serialize(), headers=headers, timeout=60)
         self.nonce = response.headers["Replay-Nonce"]
         self.debugresponse(response)
 
@@ -191,15 +210,13 @@ def notify(self, notifyurl):
         token = jwt.JWS(payload=json.dumps({}))
         token.add_signature(self.key, alg="ES256", protected=protected)
         headers = {"Content-Type": "application/jose+json"}
-        response = requests.post(
-            notifyurl, data=token.serialize(), headers=headers, timeout=60
-        )
+        response = requests.post(notifyurl, data=token.serialize(), headers=headers, timeout=60)
         self.nonce = response.headers["Replay-Nonce"]
         self.debugresponse(response)
         assert response.json()["status"] in ["pending", "valid"]
         return response.json()["status"], response.json()["url"]
 
-    def final(self, keynum, csr):
+    def final(self, keynum, csr, jwt_token: str):
         """
         There is an order, we are correct. Now we get to request a certificate.
         To do this we provide a CSR and that gets signed with the root/sub-CA
@@ -213,18 +230,18 @@ def final(self, keynum, csr):
             "kid": self.kid,
         }
         payload = {
-            #    "csr": b64encode(csr).decode().replace('+','-').replace('/','_'),
-            "csr": urlsafe_b64encode(csr)
-            .decode()
-            .rstrip("="),
+            "csr": urlsafe_b64encode(csr).decode().rstrip("="),
         }
         self.debugrequest(protected, payload)
         token = jwt.JWS(payload=json.dumps(payload))
         token.add_signature(self.key, alg="ES256", protected=protected)
-        headers = {"Content-Type": "application/jose+json"}
-        response = requests.post(
-            self.finalize[keynum], data=token.serialize(), headers=headers, timeout=60
-        )
+        headers = {
+            "Content-Type": "application/jose+json",
+            "X-Acme-Jwt": jwt_token,
+        }
+
+        # This calls the finalize method, preparing the certificate
+        response = requests.post(self.finalize[keynum], data=token.serialize(), headers=headers, timeout=60)
         self.nonce = response.headers["Replay-Nonce"]
         self.debugresponse(response)
         assert response.json()["status"] == "valid"
@@ -246,11 +263,8 @@ def getcert(self):
         token = jwt.JWS(payload="")
         token.add_signature(self.key, alg="ES256", protected=protected)
         headers = {"Content-Type": "application/jose+json"}
-        response = requests.post(
-            self.certurl, data=token.serialize(), headers=headers, timeout=60
-        )
+        response = requests.post(self.certurl, data=token.serialize(), headers=headers, timeout=60)
         self.nonce = response.headers["Replay-Nonce"]
-        self.debugresponse(response)
         return response.text
 
     def clean_headers(self, headers):
@@ -292,6 +306,4 @@ def pprint(self, data):
         """
         A simple hack to learn pprint to add some spaces upfront. Better for viewing
         """
-        print(
-            "\n".join(["    " + x for x in pprint.pformat(data, width=80).splitlines()])
-        )
+        print("\n".join(["    " + x for x in pprint.pformat(data, width=80).splitlines()]))

app/acme_directory_configuration.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class ACMEDirectoryConfiguration:
+    """
+    This data is generated from the directory endpoint. These endpoints can be different per server.
+    """
+
+    new_order_url: str
+    new_account_url: str
+    new_nonce_url: str
+    revoke_cert_url: str