WIP: check input?

vault_oidc_ssh_cert_action.py

@@ -1,7 +1,7 @@
 import os
 import subprocess
 import tempfile
-from typing import Tuple
+from typing import List, Tuple
 from urllib.parse import urlparse
 
 import requests
@@ -28,6 +28,28 @@ def _set_step_output(name: str, value: str) -> None:
         ghof.write(f"{name}={value}\n")
 
 
+def _check_inputs() -> None:
+    required_inputs = [
+        "oidc_backend_path",
+        "oidc_role",
+        "ssh_backend_path",
+        "ssh_role",
+        "vault_server",
+    ]
+    missing_inputs: List[str] = []
+    for input in required_inputs:
+        if not os.environ.get(input.upper(), "").strip():
+            missing_inputs.append(input)
+
+    if not missing_inputs:
+        return
+
+    title = "Missing Action inputs"
+    message = f"Missing required input(s): {','.join(missing_inputs)}"
+    _set_error_message(title, message)
+    raise VoscaError(title)
+
+
 def _determine_audience(input_audience: str, vault_server: str) -> str:
     if input_audience:
         return input_audience
@@ -159,6 +181,8 @@ def _revoke_token(vault_server: str, vault_token: str) -> None:
 
 
 def run() -> None:
+    _check_inputs()
+
     input_audience = os.environ["JWT_AUDIENCE"].strip()
     oidc_role = os.environ["OIDC_ROLE"].strip()
     oidc_backend = os.environ["OIDC_BACKEND_PATH"].strip("/ ")