JC Beasley
9891d7e4d5
- Updated DREAMS.md and MEMORY.md with new entries - Added daily dreaming session corpus (2026-07-05 through 2026-07-18) - Added deep/light/REM dream analysis files for July 6-19 - Added Intune rollout outputs for KMCC client - Added Vault API integration scripts and documentation - Added container update playbook (update-containers.yml) - Added .gitignore for generated/noise files
109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Generic API wrapper using Vault for dynamic token retrieval.
|
|
Supports Vault paths where tokens are stored as keys under a category.
|
|
Usage: python3 vault_api.py <METHOD> <BASE_URL> <API_PATH> <VAULT_PATH> <TOKEN_KEY> [header_type] [payload_file]
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import urllib.request
|
|
import ssl
|
|
|
|
# Disable SSL verification for self-signed certs
|
|
ssl._create_default_https_context = ssl._create_unverified_context
|
|
|
|
VAULT_URL = "https://beavault.beawit.net:8200"
|
|
|
|
def vault_auth():
|
|
"""Authenticate to Vault and return client token."""
|
|
auth_data = json.dumps({
|
|
"role_id": "75d2dcfb-9c65-7f60-59b4-eee8c7f8dc0e",
|
|
"secret_id": "6202b465-2f25-547c-ec07-f47cfc4dda3e"
|
|
}).encode()
|
|
|
|
req = urllib.request.Request(
|
|
f"{VAULT_URL}/v1/auth/approle/login",
|
|
data=auth_data,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST"
|
|
)
|
|
|
|
with urllib.request.urlopen(req) as resp:
|
|
data = json.loads(resp.read())
|
|
return data["auth"]["client_token"]
|
|
|
|
def get_service_token(vault_token, vault_path, token_key):
|
|
"""Retrieve specific token from Vault category."""
|
|
req = urllib.request.Request(
|
|
f"{VAULT_URL}/v1/kv/data/{vault_path}",
|
|
headers={"X-Vault-Token": vault_token}
|
|
)
|
|
|
|
with urllib.request.urlopen(req) as resp:
|
|
data = json.loads(resp.read())
|
|
secret_data = data["data"]["data"]
|
|
if token_key in secret_data:
|
|
return secret_data[token_key]
|
|
else:
|
|
available = list(secret_data.keys())
|
|
raise KeyError(f"Token key '{token_key}' not found. Available: {available}")
|
|
|
|
def api_call(method, base_url, api_path, vault_path, token_key, header_type="Bearer", payload_file=None):
|
|
"""Make authenticated API call."""
|
|
vault_token = vault_auth()
|
|
service_token = get_service_token(vault_token, vault_path, token_key)
|
|
|
|
url = f"{base_url}{api_path}"
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
if header_type.lower() == "xc-token":
|
|
headers["xc-token"] = service_token
|
|
elif header_type.lower() in ["api-key", "x-api-key"]:
|
|
headers["api-key"] = service_token
|
|
elif header_type.lower() == "x-vault-token":
|
|
headers["X-Vault-Token"] = service_token
|
|
else:
|
|
headers["Authorization"] = f"Bearer {service_token}"
|
|
|
|
if method == "GET":
|
|
req = urllib.request.Request(url, headers=headers)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read())
|
|
elif method in ["POST", "PATCH"]:
|
|
data = None
|
|
if payload_file:
|
|
import os
|
|
if os.path.exists(payload_file):
|
|
with open(payload_file) as f:
|
|
data = json.dumps(json.load(f)).encode()
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read())
|
|
else:
|
|
raise ValueError(f"Unsupported method: {method}")
|
|
|
|
def main():
|
|
if len(sys.argv) < 6:
|
|
print("Usage: python3 vault_api.py <METHOD> <BASE_URL> <API_PATH> <VAULT_PATH> <TOKEN_KEY> [header_type] [payload_file]")
|
|
print("Examples:")
|
|
print(' python3 vault_api.py GET "https://grist.beawit.net" "/api/..." "api/integration" "grist"')
|
|
print(' python3 vault_api.py GET "http://192.168.25.5:8080" "/api/..." "api/infrastructure" "nocodb-token" "xc-token"')
|
|
print(' python3 vault_api.py GET "http://192.168.19.17:6333" "/collections" "api/infrastructure" "qdrant-api-key" "api-key"')
|
|
sys.exit(1)
|
|
|
|
method = sys.argv[1].upper()
|
|
base_url = sys.argv[2]
|
|
api_path = sys.argv[3]
|
|
vault_path = sys.argv[4]
|
|
token_key = sys.argv[5]
|
|
header_type = sys.argv[6] if len(sys.argv) > 6 else "Bearer"
|
|
payload_file = sys.argv[7] if len(sys.argv) > 7 else None
|
|
|
|
result = api_call(method, base_url, api_path, vault_path, token_key, header_type, payload_file)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|