feat(add $VAR expansion from config):

agentic_security/core/app.py

@@ -1,3 +1,4 @@
+import os
 from asyncio import Event, Queue
 
 from fastapi import FastAPI
@@ -5,6 +6,7 @@ from fastapi import FastAPI
 tools_inbox: Queue = Queue()
 stop_event: Event = Event()
 current_run: str = {"spec": "", "id": ""}
+_secrets = {}
 
 
 def create_app() -> FastAPI:
@@ -33,3 +35,20 @@ def set_current_run(spec):
     current_run["id"] = hash(id(spec))
     current_run["spec"] = spec
     return current_run
+
+
+def get_secrets():
+    return _secrets
+
+
+def set_secrets(secrets):
+    _secrets.update(secrets)
+    expand_secrets(_secrets)
+    return _secrets
+
+
+def expand_secrets(secrets):
+    for key in secrets:
+        val = secrets[key]
+        if val.startswith("$"):
+            secrets[key] = os.getenv(val.strip("$"))

agentic_security/core/test_app.py

@@ -0,0 +1,27 @@
+import os
+import pytest
+from agentic_security.core.app import expand_secrets
+
+
+@pytest.fixture(autouse=True)
+def setup_env_vars():
+    # Set up environment variables for testing
+    os.environ["TEST_ENV_VAR"] = "test_value"
+
+
+def test_expand_secrets_with_env_var():
+    secrets = {"secret_key": "$TEST_ENV_VAR"}
+    expand_secrets(secrets)
+    assert secrets["secret_key"] == "test_value"
+
+
+def test_expand_secrets_without_env_var():
+    secrets = {"secret_key": "$NON_EXISTENT_VAR"}
+    expand_secrets(secrets)
+    assert secrets["secret_key"] is None
+
+
+def test_expand_secrets_without_dollar_sign():
+    secrets = {"secret_key": "plain_value"}
+    expand_secrets(secrets)
+    assert secrets["secret_key"] == "plain_value"