Add Public EDI TOKEN for all public requests (#13)

Author: servilla

tests/test_iam.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+:Mod:
+    test_iam
+
+:Synopsis:
+
+:Author:
+    pasta
+
+:Created:
+    7/31/25
+"""
+import pytest
+
+from config import Config
+from edi.iam import IAM
+
+
+@pytest.mark.asyncio
+async def test_create_token():
+    iam = IAM()
+    edi_token = await iam.create_token(Config.PUBLIC_ID)
+    assert edi_token is not None

webapp/auth/authenticate.py

@@ -30,6 +30,7 @@
 from auth.pasta_token import PastaToken
 import auth.pasta_crypto as pasta_crypto
 from config import Config
+from edi.iam import IAM
 
 
 logger = daiquiri.getLogger(__name__)
@@ -38,6 +39,7 @@
 async def authenticate(request: Request) -> tuple:
     pasta_token = PastaToken()
     edi_token = None
+    is_public = False;
 
     # Old-style PASTA authentication
     if "authorization" in request.headers:
@@ -60,6 +62,8 @@ async def authenticate(request: Request) -> tuple:
     else:
         pasta_token.uid = Config.PUBLIC
         pasta_token.system = Config.SYSTEM
+        is_public = True
+
     msg = f"Authentication for user: '{pasta_token.to_string()}'"
     logger.info(msg)
 
@@ -68,6 +72,9 @@ async def authenticate(request: Request) -> tuple:
         edi_token = request.cookies.get("edi-token")
         msg = f"EDI Token '{edi_token}' exists"
         logger.info(msg)
+    elif is_public:
+        iam = IAM()
+        edi_token = await iam.create_token(Config.PUBLIC_ID)
 
     return pasta_token, edi_token
 

webapp/edi/__init__.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+:Mod:
+    __init__.py
+
+:Synopsis:
+
+:Author:
+    pasta
+
+:Created:
+    7/31/25
+"""

webapp/edi/iam.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+:Mod:
+    iam
+
+:Synopsis:
+
+:Author:
+    Mark Servilla
+
+:Created:
+    7/31/25
+"""
+import json
+from pathlib import Path
+
+import daiquiri
+import httpx
+import ssl
+
+from config import Config
+
+
+logger = daiquiri.getLogger(__name__)
+
+
+
+class IAM:
+
+    def __init__(self):
+        self.base_url = Config.AUTH
+
+    async def create_token(self, edi_id: str) -> str:
+        route = f"/auth/v1/token/{edi_id}"
+        url = self.base_url + route
+        data = {
+            "key": Config.AUTH_KEY
+        }
+
+        verify = True
+        if Path(str(Config.CA_FILE)).exists() and Path(str(Config.CA_FILE)).is_file():
+            # Create local SSL CA context if Config.CA_FILE is valid path
+            verify = ssl.create_default_context(cafile=Config.CA_FILE)
+        else:
+            msg = f"Truststore file '{Config.CA_FILE}' does not exist"
+            logger.error(msg)
+        try:
+            response = httpx.post(url, json=data, verify=verify)
+            response.raise_for_status()
+        except httpx.HTTPError as ex:
+            logger.error(ex)
+            raise ex
+
+        payload = response.json()
+        try:
+            edi_token = payload["token"]
+        except KeyError as ex:
+            logger.error(ex)
+            edi_token = None
+
+        return edi_token