refact: support secrets from env vars and include vault namespace header

app/permissions_engine/vault.rego

@@ -17,6 +17,8 @@ test := "test" if {
 
 else := "opa"
 
+# TODO: need to pass the 'X-Vault-Namespace' header to the vault requests
+
 # paths are the paths authorized for methods, used by permissions.rego
 paths := http.send({"method": "get", "url": concat("/", ["VAULT_URL/v1", test, "paths"]), "headers": {"X-Vault-Token": vault_token}}).body.data.paths
 

app/refresh_stores.py

@@ -1,10 +1,9 @@
 import json
 import os
-from src.auth import get_vault_token_for_service, reload_comanage
+from src.auth import get_secret_file_value_or_env, get_vault_token_for_service, reload_comanage
 import sys
 import requests
 
-
 # get the token for the opa store
 try:
     with open("/app/permissions_engine/opa_secret.json") as f:
@@ -13,16 +12,18 @@
             "X-Opa": opa_json["opa_secret"],
             "Content-Type": "application/json; charset=utf-8"
         }
-        with open("/home/pcgl/opa-roleid") as f:
-            role_id = f.read().strip()
-            payload = f"{{\"token\": \"{get_vault_token_for_service("opa", role_id=role_id)}\"}}"
-            response = requests.put(url=f"{os.getenv('OPA_URL')}/v1/data/opa_token", headers=headers, data=payload)
-            print(response.text)
-        with open("/home/pcgl/test-roleid") as f:
-            role_id = f.read().strip()
-            payload = f"{{\"token\": \"{get_vault_token_for_service("test", role_id=role_id)}\"}}"
-            response = requests.put(url=f"{os.getenv('OPA_URL')}/v1/data/test_token", headers=headers, data=payload)
-            print(response.text)
+
+        # OPA Role-ID
+        opa_role_id = get_secret_file_value_or_env("/home/pcgl/opa-roleid", "OPA_ROLEID")
+        payload = f"{{\"token\": \"{get_vault_token_for_service("opa", role_id=opa_role_id)}\"}}"
+        response = requests.put(url=f"{os.getenv('OPA_URL')}/v1/data/opa_token", headers=headers, data=payload)
+        print(response.text)
+
+        # Test Role-ID
+        test_role_id = get_secret_file_value_or_env("/home/pcgl/test-roleid", "TEST_ROLEID")
+        payload = f"{{\"token\": \"{get_vault_token_for_service("test", role_id=test_role_id)}\"}}"
+        response = requests.put(url=f"{os.getenv('OPA_URL')}/v1/data/test_token", headers=headers, data=payload)
+        print(response.text)
 
     print(reload_comanage())
 except Exception as e:

app/src/auth.py

@@ -9,6 +9,7 @@
 ## Env vars for most auth methods:
 OPA_URL = os.getenv('OPA_URL', "http://localhost:8181")
 VAULT_URL = os.getenv('VAULT_URL', "http://localhost:8200")
+VAULT_NAMESPACE = os.getenv("VAULT_NAMESPACE", "")
 SERVICE_NAME = os.getenv("SERVICE_NAME")
 APPROLE_TOKEN_FILE = os.getenv("APPROLE_TOKEN_FILE", "/home/pcgl/approle-token")
 VERIFY_ROLE_ID_FILE = "/home/pcgl/verify-roleid"
@@ -114,6 +115,41 @@ def exchange_refresh_token(refresh_token, client_id=PCGL_CLIENT_ID, client_secre
     raise AuthzError(response.text)
 
 
+def get_secret_file_value_or_env(file_path: str, env_var: str) -> str:
+    """
+    Tries to reads and return the value contained at `file_path` if it exists.\n
+    Otherwise tries to read the value from the `env_var` environment variable if it exists.\n
+    Raises an `AuthzError` if neither can be found.
+    """
+    if file_path and os.path.exists(file_path):
+        with open(file_path, 'r') as f:
+            secret = f.read().strip()
+        return secret
+    if env_var:
+        return os.getenv(env_var)
+    raise AuthzError(f"Couldn't read the secret from the {file_path} path or the {env_var} env variable.")
+
+
+def get_vault_namespace_header() -> dict:
+    if VAULT_NAMESPACE:
+        return {
+            "X-Vault-Namespace": VAULT_NAMESPACE
+        }
+    return {}
+
+
+def get_vault_headers(token: str) -> dict:
+    """
+    Returns authentication headers for Vault API HTTP requests.\n
+    The `token` argument is passed to the `X-Vault-Token` header.\n
+    If the `VAULT_NAMESPACE` env var is set, includes its value in the `X-Vault-Namespace` header.
+    """
+    return {
+        "X-Vault-Token": token,
+        **get_vault_namespace_header()
+    }
+
+
 ######
 # General authorization methods
 ######
@@ -520,16 +556,14 @@ def get_vault_token_for_service(service=SERVICE_NAME, approle_token=None, role_i
         raise AuthzError("no SERVICE_NAME specified")
     # in CanDIGv2 docker stack, approle token should have been passed in
     if approle_token is None:
-        with open(APPROLE_TOKEN_FILE) as f:
-            approle_token = f.read().strip()
+        approle_token = get_secret_file_value_or_env(APPROLE_TOKEN_FILE, "APPROLE_TOKEN")
     if approle_token is None:
         raise AuthzError("no approle token found")
 
     # in CanDIGv2 docker stack, roleid should have been passed in
     if role_id is None:
         try:
-            with open(f"/home/pcgl/{service}-roleid") as f:
-                role_id = f.read().strip()
+            role_id = get_secret_file_value_or_env(f"/home/pcgl/{service}-roleid", f"{service.upper()}-ROLEID")
         except Exception as e:
             raise AuthzError(str(e))
     if role_id is None:
@@ -538,7 +572,7 @@ def get_vault_token_for_service(service=SERVICE_NAME, approle_token=None, role_i
     # get the secret_id
     if secret_id is None:
         url = f"{VAULT_URL}/v1/auth/approle/role/{service}/secret-id"
-        headers = { "X-Vault-Token": approle_token }
+        headers = get_vault_headers(token=approle_token)
         response = requests.post(url=url, headers=headers)
         if response.status_code == 200:
             secret_id = response.json()["data"]["secret_id"]
@@ -573,9 +607,7 @@ def set_service_store_secret(service=SERVICE_NAME, key=None, value=None, role_id
     if key is None:
         return {"error": "no key specified"}, 400
 
-    headers = {
-        "X-Vault-Token": token
-    }
+    headers = get_vault_headers(token=token)
     url = f"{VAULT_URL}/v1/{service}/{key}"
     if "dict" in str(type(value)):
         value = json.dumps(value)
@@ -599,9 +631,7 @@ def get_service_store_secret(service=SERVICE_NAME, key=None, role_id=None, secre
     if key is None:
         return {"error": "no key specified"}, 400
 
-    headers = {
-        "X-Vault-Token": token
-    }
+    headers = get_vault_headers(token)
     url = f"{VAULT_URL}/v1/{service}/{key}"
     response = requests.get(url, headers=headers)
     if response.status_code == 200:
@@ -624,9 +654,7 @@ def delete_service_store_secret(service=SERVICE_NAME, key=None, role_id=None, se
     if key is None:
         return {"error": "no key specified"}, 400
 
-    headers = {
-        "X-Vault-Token": token
-    }
+    headers = get_vault_headers(token)
     url = f"{VAULT_URL}/v1/{service}/{key}"
     response = requests.delete(url, headers=headers)
     return response.text, response.status_code
@@ -639,16 +667,13 @@ def create_service_token(service_uuid):
     # this will get us a fresh approle token for this service
     role_id = None
     try:
-        with open(VERIFY_ROLE_ID_FILE) as f:
-            role_id = f.read().strip()
+        role_id = get_secret_file_value_or_env(VERIFY_ROLE_ID_FILE, "VERIFY_ROLE_ID")
     except Exception as e:
         raise AuthzError(str(e))
 
     token = get_vault_token_for_service("verify", role_id=role_id)
 
-    headers = {
-        "X-Vault-Token": token
-    }
+    headers = get_vault_headers(token)
 
     # create the random service-token:
     url = f"{VAULT_URL}/v1/cubbyhole/{token}"