Merge branch 'uzipoc_q4_2024' into feature/directory-usage

.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

.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
+

app/acme.py

@@ -45,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):
         """
@@ -154,9 +152,7 @@ def challenge(self, challengeurl):
         # 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
-        )
+        response = requests.post(challengeurl, data=token.serialize(), headers=headers, timeout=60)
         returned_json = response.json()
 
         self.debugresponse(response)
@@ -192,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)
 
@@ -216,9 +210,7 @@ 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"]
@@ -249,9 +241,7 @@ def final(self, keynum, csr, jwt_token: str):
         }
 
         # This calls the finalize method, preparing the certificate
-        response = requests.post(
-            self.finalize[keynum], data=token.serialize(), headers=headers, timeout=60
-        )
+        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"
@@ -273,9 +263,7 @@ 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"]
         return response.text
 
@@ -318,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/page/creatersakey.py

@@ -16,9 +16,7 @@ def __init__(self, mypkcs, parent=None):
 
         layout = QVBoxLayout(self)
 
-        self.emptyWarningCheckbox = QCheckBox(
-            "I understand that the YubiKey will be emptied"
-        )
+        self.emptyWarningCheckbox = QCheckBox("I understand that the YubiKey will be emptied")
         self.emptyWarningCheckbox.setStyleSheet("color: red")
         self.emptyWarningCheckbox.toggled.connect(self.updateNextButtonStatus)
         layout.addWidget(self.emptyWarningCheckbox)
@@ -76,9 +74,7 @@ def updateProgress(self):
                 self.stepsCompleted = True
                 self.completeKeyCreationProcess()
             return
-        self.progressLabel.setText(
-            f"Creating key {self.currentStep} of {self.totalSteps}..."
-        )
+        self.progressLabel.setText(f"Creating key {self.currentStep} of {self.totalSteps}...")
         selectedYubiKeySlot, _, _ = self.wizard().property("selectedYubiKey")
         worker = Worker(self.pkcs, self.currentStep, selectedYubiKeySlot)
         worker.finished.connect(self.finishCurrentStep)
@@ -152,23 +148,14 @@ def checkIfYubiKeyFilled(self, _selectedYubiKey):
                     if label == HEADERS[col] + " for " + x and col < 3:
                         finds[col][row] = True
                         break
-                    if (
-                        label == "X.509 Certificate for PIV Attestation" + y
-                        and col == 3
-                    ):
+                    if label == "X.509 Certificate for PIV Attestation" + y and col == 3:
                         finds[col][row] = True
                         break
         self.pkcs.delsession(selectedYubiKeySlot)
         print(finds)
-        return any(
-            value for inner_dict in finds.values() for value in inner_dict.values()
-        )
+        return any(value for inner_dict in finds.values() for value in inner_dict.values())
 
     def updateNextButtonStatus(self):
-        yubiKeyFilled = self.checkIfYubiKeyFilled(
-            self.wizard().property("selectedYubiKey")
-        )
+        yubiKeyFilled = self.checkIfYubiKeyFilled(self.wizard().property("selectedYubiKey"))
         checkboxChecked = self.emptyWarningCheckbox.isChecked()
-        self.wizard().button(QWizard.WizardButton.NextButton).setEnabled(
-            not yubiKeyFilled or checkboxChecked
-        )
+        self.wizard().button(QWizard.WizardButton.NextButton).setEnabled(not yubiKeyFilled or checkboxChecked)

app/page/logindigid.py

@@ -35,9 +35,7 @@ def __init__(
         self._oidc_provider_base_url = oidc_provider_base_url
 
     def _get_jwt_url(self):
-        return urllib.parse.urljoin(
-            self._oidc_provider_base_url.geturl(), "ziekenboeg/users/jwt"
-        )
+        return urllib.parse.urljoin(self._oidc_provider_base_url.geturl(), "ziekenboeg/users/jwt")
 
     def initializePage(self):
         layout = QVBoxLayout(self)
@@ -56,15 +54,11 @@ def initializePage(self):
             self.acme.order(keynum)
             self.acme.getchallenge(keynum - 1)
 
-        login_url = urllib.parse.urljoin(
-            self._oidc_provider_base_url.geturl(), "oidc/login"
-        )
+        login_url = urllib.parse.urljoin(self._oidc_provider_base_url.geturl(), "oidc/login")
         url = QUrl(login_url)
 
         query = QUrlQuery()
-        query.addQueryItem(
-            "acme_tokens", ",".join(self.acme.tokens)
-        )  # Replace with your parameter name and value
+        query.addQueryItem("acme_tokens", ",".join(self.acme.tokens))  # Replace with your parameter name and value
         url.setQuery(query)
         self.browser.load(url)
 
@@ -81,9 +75,7 @@ def isComplete(self):
     def onUrlChanged(self, url):
         print(url.toString())
 
-        user_home_url = urllib.parse.urljoin(
-            self._oidc_provider_base_url.geturl(), "ziekenboeg/users/home"
-        )
+        user_home_url = urllib.parse.urljoin(self._oidc_provider_base_url.geturl(), "ziekenboeg/users/home")
         if url.toString() == user_home_url:
             self.browser.load(QUrl(self._get_jwt_url()))
 

app/page/selectkey.py

@@ -28,7 +28,7 @@ def __init__(self, mypkcs, parent=None):
             try:
                 info = self.pkcs.pkcs11.getTokenInfo(slot)
                 yubikeys += [(info.model, info.serialNumber, info.label, slot)]
-            except:
+            except Exception:
                 pass
 
         for name, serial, available, slot in yubikeys:
@@ -45,9 +45,7 @@ def __init__(self, mypkcs, parent=None):
     def yubiKeySelected(self, current, _previous):
         if current is not None:
             # Assuming the YubiKeyItemWidget has a method or property to get the YubiKey details
-            self.selectedYubiKey = self.yubikeyListWidget.itemWidget(
-                current
-            ).getYubiKeyDetails()
+            self.selectedYubiKey = self.yubikeyListWidget.itemWidget(current).getYubiKeyDetails()
             self.wizard().button(QWizard.WizardButton.NextButton).setEnabled(True)
 
     def nextId(self):

app/page/worker.py

@@ -15,7 +15,7 @@ def __init__(self, mypkcs, currentStep, selectedYubiKeySlot):
     def task(self):
         try:
             print("** Worker", self.currentStep)
-            # keysize of 4096 is not supported, for more info: 
+            # keysize of 4096 is not supported, for more info:
             # https://github.com/Yubico/yubico-piv-tool/issues/58
             public_template = [
                 (PyKCS11.CKA_CLASS, PyKCS11.CKO_PUBLIC_KEY),
@@ -35,12 +35,10 @@ def task(self):
                 (PyKCS11.CKA_ID, [self.currentStep]),
             ]
             session = self.pkcs.getadminsession(self.selectedYubiKeySlot)
-            print(
-                "** Output", session.generateKeyPair(public_template, private_template)
-            )
+            print("** Output", session.generateKeyPair(public_template, private_template))
             self.pkcs.delsession(self.selectedYubiKeySlot)
             # print("done")
-        except:
+        except Exception:
             print("Bla")
         finally:
             print("** Worker done", self.currentStep)

app/pkcs.py

@@ -73,9 +73,7 @@ def delsession(self, slot):
 
     def listattest(self, slot):
         session = self.getsession(slot)
-        all_objects = session.findObjects(
-            [(PyKCS11.CKA_CLASS, PyKCS11.CKO_CERTIFICATE)]
-        )
+        all_objects = session.findObjects([(PyKCS11.CKA_CLASS, PyKCS11.CKO_CERTIFICATE)])
         for obj in all_objects:
             label = session.getAttributeValue(obj, [PyKCS11.CKA_LABEL])[0]
             value = session.getAttributeValue(obj, [PyKCS11.CKA_VALUE])[0]
@@ -110,9 +108,7 @@ def getattest(self, slot, num):
 
     def listprivatekeys(self, slot):
         session = self.getsession(slot)
-        all_objects = session.findObjects(
-            [(PyKCS11.CKA_CLASS, PyKCS11.CKO_PRIVATE_KEY)]
-        )
+        all_objects = session.findObjects([(PyKCS11.CKA_CLASS, PyKCS11.CKO_PRIVATE_KEY)])
         print(all_objects)
         self.delsession(slot)
 
@@ -180,15 +176,9 @@ def makecsr(self, session, private_key, public_key):
                     },
                     "public_key": RSAPublicKey(
                         {
-                            "modulus": int.from_bytes(
-                                session.getAttributeValue(
-                                    public_key, [PyKCS11.CKA_MODULUS]
-                                )[0]
-                            ),
+                            "modulus": int.from_bytes(session.getAttributeValue(public_key, [PyKCS11.CKA_MODULUS])[0]),
                             "public_exponent": int.from_bytes(
-                                session.getAttributeValue(
-                                    public_key, [PyKCS11.CKA_PUBLIC_EXPONENT]
-                                )[0]
+                                session.getAttributeValue(public_key, [PyKCS11.CKA_PUBLIC_EXPONENT])[0]
                             ),
                         }
                     ),
@@ -218,12 +208,8 @@ def makecsr(self, session, private_key, public_key):
     def getcsr(self, slot, keyid):
         csr = None
         session = self.getusersession(slot)
-        privkey = session.findObjects(
-            [(PyKCS11.CKA_CLASS, PyKCS11.CKO_PRIVATE_KEY), (PyKCS11.CKA_ID, [keyid])]
-        )
-        pubkey = session.findObjects(
-            [(PyKCS11.CKA_CLASS, PyKCS11.CKO_PUBLIC_KEY), (PyKCS11.CKA_ID, [keyid])]
-        )
+        privkey = session.findObjects([(PyKCS11.CKA_CLASS, PyKCS11.CKO_PRIVATE_KEY), (PyKCS11.CKA_ID, [keyid])])
+        pubkey = session.findObjects([(PyKCS11.CKA_CLASS, PyKCS11.CKO_PUBLIC_KEY), (PyKCS11.CKA_ID, [keyid])])
         if privkey and pubkey:
             privkey = privkey[0]
             pubkey = pubkey[0]

app/pkcs_lib_finder.py

@@ -15,9 +15,7 @@ class PKCS11LibFinder:
     _LINUX_X86_DEFAULT_PATH = pathlib.Path("/usr/lib/libykcs11.so")
     _LINUX_X64_DEFAULT_PATH = pathlib.Path("/usr/lib64/libykcs11.so")
 
-    _WINDOWS_DEFAULT_LOCATION = pathlib.Path(
-        "C:\\Program Files\\Yubico\\YubiKey PIV Tool\\ykcs11.dll"
-    )
+    _WINDOWS_DEFAULT_LOCATION = pathlib.Path("C:\\Program Files\\Yubico\\YubiKey PIV Tool\\ykcs11.dll")
     _WINDOWS_DEFAULT_SYSTEM32_PATH = pathlib.Path("C:\\Windows\\System32\\ykcs11.dll")
     _WINDOWS_DEFAULT_SYSWOW64_PATH = pathlib.Path("C:\\Windows\\SysWOW64\\ykcs11.dll")
 
@@ -37,9 +35,7 @@ def _try_default_location(self, libpath: pathlib.Path):
 
         return lib
 
-    def _try_load_from_paths(
-        self, paths: list[pathlib.Path]
-    ) -> Optional[PyKCS11.PyKCS11Lib]:
+    def _try_load_from_paths(self, paths: list[pathlib.Path]) -> Optional[PyKCS11.PyKCS11Lib]:
         for path in paths:
             try:
                 logger.info(f"Trying to parse libykcs11 library from {path}...")

app/wizard.py

@@ -24,27 +24,21 @@
 import urllib.parse
 from dotenv import load_dotenv
 
-DEFAULT_ACME_CA_SERVER_URL = (
-    "https://acme.proeftuin.uzi-online.irealisatie.nl/directory"
-)
+DEFAULT_ACME_CA_SERVER_URL = "https://acme.proeftuin.uzi-online.irealisatie.nl/directory"
 DEFAULT_YUBIKEY_PIN = "123456"
 DEFAULT_PROEFTUIN_OIDC_LOGIN_URL = "https://proeftuin.uzi-online.irealisatie.nl"
 
 
 class MainWindow(QMainWindow):
-    def __init__(
-        self, mypkcs, myacme, oidc_provider_base_url: urllib.parse.ParseResult
-    ):
+    def __init__(self, mypkcs, myacme, oidc_provider_base_url: urllib.parse.ParseResult):
         super().__init__()
 
         self.setWindowTitle("YubiKey Wizard")
         self.resize(1024, 768)
 
         # Create the wizard
         self.wizard = QWizard()
-        self.wizard.setSizePolicy(
-            QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding
-        )
+        self.wizard.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
         self.wizard.addPage(WelcomePage())
         self.wizard.addPage(SelectYubiKeyPage(mypkcs))
         self.wizard.addPage(CreateRSAKeysPage(mypkcs))
@@ -69,12 +63,8 @@ def __init__(
     pkcslib = PKCS11LibFinder().find()
     pkcscls = pkcs(pykcs11lib=pkcslib, yubikey_pin=yubikey_pin)
 
-    oidc_provider_url = urllib.parse.urlparse(
-        getenv("OIDC_PROVIDER_BASE_URL", DEFAULT_PROEFTUIN_OIDC_LOGIN_URL)
-    )
-    acme_ca_server_url = urllib.parse.urlparse(
-        getenv("ACME_SERVER_DIRECTORY_URL", DEFAULT_ACME_CA_SERVER_URL)
-    )
+    oidc_provider_url = urllib.parse.urlparse(getenv("OIDC_PROVIDER_BASE_URL", DEFAULT_PROEFTUIN_OIDC_LOGIN_URL))
+    acme_ca_server_url = urllib.parse.urlparse(getenv("ACME_SERVER_DIRECTORY_URL", DEFAULT_ACME_CA_SERVER_URL))
     directory_config = ACMEDirectoryConfigurationParser().parse(acme_ca_server_url)
     acme = ACME(directory_config)
 

requirements-dev.txt

@@ -0,0 +1 @@
+ruff==0.8.1

requirements.txt

@@ -98,4 +98,4 @@ werkzeug==3.0.6
     # via flask
 wrapt==1.15.0
     # via astroid
-python-dotenv==1.0.1
+python-dotenv==1.0.1