certificate renewal

api/kiwi_vpn_api/easyrsa.py

@@ -195,7 +195,7 @@ class EasyRSA:
         cert_filename: Path,
         *easyrsa_cmd: str,
         **easyrsa_env: str,
-    ) -> x509.Certificate:
+    ) -> x509.Certificate | None:
         """
         Create an X.509 certificate
         """
@@ -213,6 +213,7 @@ class EasyRSA:
         if (cert_expiry_days := config.crypto.cert_expiry_days) is not None:
             easyrsa_env["EASYRSA_CERT_EXPIRE"] = str(cert_expiry_days)
 
+        try:
             # call easyrsa
             self.__easyrsa(
                 *easyrsa_cmd,
@@ -234,6 +235,10 @@ class EasyRSA:
                     cert_file.read()
                 )
 
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            # certificate couldn't be built
+            return None
+
     def init_pki(self) -> None:
         """
         Clean working directory
@@ -254,6 +259,8 @@ class EasyRSA:
             EASYRSA_REQ_CN="kiwi-vpn-ca",
         )
 
+        assert cert is not None
+
         # # this takes long!
         # self.__easyrsa("gen-dh")
         return cert
@@ -284,6 +291,27 @@ class EasyRSA:
             **dn.easyrsa_env,
         )
 
+    def renew(
+        self,
+        dn: DistinguishedName | None = None,
+    ) -> x509.Certificate | None:
+        """
+        Issue a client or server certificate
+        """
+
+        if dn is None:
+            dn = DistinguishedName.build()
+
+        return self.__build_cert(
+            Path("issued").joinpath(f"{dn.common_name}.crt"),
+
+            "renew",
+            dn.common_name,
+            "nopass",
+
+            **dn.easyrsa_env,
+        )
+
 
 EASYRSA = EasyRSA()
 

api/kiwi_vpn_api/routers/device.py

@@ -119,3 +119,46 @@ async def request_certificate_issuance(
     # return updated device
     device.update()
     return device
+
+
+@router.post(
+    "/{device_id}/renew",
+    responses={
+        status.HTTP_200_OK: Responses.OK,
+        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
+        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
+        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
+    },
+    response_model=DeviceRead,
+)
+async def request_certificate_renewal(
+    current_user: User = Depends(get_current_user),
+    device: Device = Depends(get_device_by_id),
+) -> Device:
+    """
+    POST ./{device_id}/renew: Request certificate renewal for a device.
+    """
+
+    # check permission
+    if not current_user.can_edit(device):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+    # can only renew an already certified device
+    if device.approved is not True:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+
+    # check if we must wait for approval
+    device.approved = current_user.can_renew
+
+    if device.approved:
+        # renew the certificate immediately
+        if (certificate := EASYRSA.renew(
+            dn=DistinguishedName.build(device)
+        )) is not None:
+            device.expiry = certificate.not_valid_after
+
+    # return updated device
+    device.update()
+    return device