device revoke endpoint

api/kiwi_vpn_api/easyrsa.py

@@ -182,8 +182,15 @@ class EasyRSA:
                 *easyrsa_cmd,
             ],
             env={
+                # base settings
                 "EASYRSA_BATCH": "1",
                 "EASYRSA_PKI": str(self.output_directory),
+
+                # always include CA password
+                "EASYRSA_PASSOUT": f"pass:{self.ca_password}",
+                "EASYRSA_PASSIN": f"pass:{self.ca_password}",
+
+                # include env from parameters
                 **easyrsa_env,
             },
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
@@ -218,10 +225,6 @@ class EasyRSA:
             self.__easyrsa(
                 *easyrsa_cmd,
 
-                # include CA password
-                EASYRSA_PASSOUT=f"pass:{self.ca_password}",
-                EASYRSA_PASSIN=f"pass:{self.ca_password}",
-
                 # include algorithm options
                 **EasyRSA.__mapKeyAlgorithm[algorithm],
                 **easyrsa_env,
@@ -312,6 +315,30 @@ class EasyRSA:
             **dn.easyrsa_env,
         )
 
+    def revoke(
+        self,
+        dn: DistinguishedName | None = None,
+    ) -> bool:
+        """
+        Revoke a client or server certificate
+        """
+
+        if dn is None:
+            dn = DistinguishedName.build()
+
+        try:
+            self.__easyrsa(
+                "revoke",
+                dn.common_name,
+
+                **dn.easyrsa_env,
+            )
+
+        except subprocess.CalledProcessError:
+            return False
+
+        return True
+
 
 EASYRSA = EasyRSA()
 

api/kiwi_vpn_api/routers/device.py

@@ -162,3 +162,43 @@ async def request_certificate_renewal(
     # return updated device
     device.update()
     return device
+
+
+@router.post(
+    "/{device_id}/revoke",
+    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 revoke_certificate(
+    current_user: User = Depends(get_current_user),
+    device: Device = Depends(get_device_by_id),
+) -> Device:
+    """
+    POST ./{device_id}/revoke: Revoke a device certificate.
+    """
+
+    # check permission
+    if not current_user.can_edit(device):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+    # can only revoke a currently certified device
+    if device.approved is not True:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+
+    # revoke the device certificate
+    EASYRSA.revoke(dn=DistinguishedName.build(device))
+
+    # reset the device
+    device.approved = None
+    device.expiry = None
+
+    # return updated device
+    device.update()
+    return device