use basic auth, if no JWT token given

.github/workflows/build_and_test.yaml

@@ -96,6 +96,7 @@ jobs:
 
           wait_for_pod_ready "sidecar"
           wait_for_pod_ready "sidecar-basicauth-args"
+          wait_for_pod_ready "sidecar-jwt-token"
           wait_for_pod_ready "sidecar-5xx"
           wait_for_pod_ready "sidecar-pythonscript"
           wait_for_pod_ready "sidecar-pythonscript-logfile"
@@ -115,7 +116,7 @@ jobs:
           sleep 20
           echo "Installing resources..."
           kubectl apply -f "test/resources/resources.yaml"
-          pods=("sidecar" "sidecar-basicauth-args" "sidecar-5xx" "sidecar-pythonscript" "sidecar-pythonscript-logfile")
+          pods=("sidecar" "sidecar-basicauth-args" "sidecar-jwt-token" "sidecar-5xx" "sidecar-pythonscript" "sidecar-pythonscript-logfile")
           resources=("sample-configmap" "sample-secret-binary" "absolute-configmap" "relative-configmap" "change-dir-configmap" "similar-configmap-secret" "url-configmap-500" "url-configmap-basic-auth" "sample-configmap")
           for p in ${pods[*]}; do
             for r in ${resources[*]}; do
@@ -140,6 +141,7 @@ jobs:
           mkdir /tmp/logs
           kubectl logs sidecar > /tmp/logs/sidecar.log
           kubectl logs sidecar-basicauth-args > /tmp/logs/sidecar-basicauth-args.log
+          kubectl logs sidecar-jwt-token > /tmp/logs/sidecar-jwt-token.log
           kubectl logs sidecar-5xx > /tmp/logs/sidecar-5xx.log
           kubectl logs sidecar-pythonscript > /tmp/logs/sidecar-pythonscript.log
           kubectl logs sidecar-pythonscript-logfile > /tmp/logs/sidecar-pythonscript-logfile.log

src/helpers.py

@@ -10,7 +10,10 @@
 import requests
 from requests.adapters import HTTPAdapter
 from requests.auth import HTTPBasicAuth
-from requests.packages.urllib3.util.retry import Retry
+from requests.adapters import Retry
+
+import jwt
+from jwt import InvalidTokenError
 
 from logger import get_logger
 import argparse
@@ -148,10 +151,26 @@ def request(url, method, enable_5xx=False, payload=None):
     username,password = fetch_basic_auth_credentials()
     encoding = 'latin1' if not os.getenv("REQ_BASIC_AUTH_ENCODING") else os.getenv("REQ_BASIC_AUTH_ENCODING")
     if username and password:
-        auth = HTTPBasicAuth(username.encode(encoding), password.encode(encoding))
+        basic_auth = HTTPBasicAuth(username.encode(encoding), password.encode(encoding))
     else:
-        auth = None
-
+        basic_auth = None
+
+    # JWT Auth
+    jwt_token = os.getenv("JWT_TOKEN")
+    jwt_header_name = os.getenv("JWT_HEADER_NAME", "X-JWT-Assertion")
+    jwt_auth = None
+    if jwt_token:
+        # Validate JWT token
+        try:
+            # We can add custom validation, such as a secret or public key
+            jwt.decode(jwt_token, options={"verify_signature": False})  # Disable signature check for simplicity    
+            logger.info("JWT Token valid:")
+            jwt_auth = {jwt_header_name: jwt_token}
+        except InvalidTokenError as e:
+            logger.error(f"Invalid JWT Token: {e}")
+            jwt_auth = None
+
+    # Create the session
     r = requests.Session()
 
     retries = Retry(total=REQ_RETRY_TOTAL,
@@ -162,15 +181,33 @@ def request(url, method, enable_5xx=False, payload=None):
                     status_forcelist=enforce_status_codes)
     r.mount("http://", HTTPAdapter(max_retries=retries))
     r.mount("https://", HTTPAdapter(max_retries=retries))
+
     if url is None:
         logger.warning(f"No url provided. Doing nothing.")
         return
 
+    # Handle Authentication logic: Basic Auth, JWT or both
+    auth = None
+    headers = {}
+
+    if basic_auth and jwt_auth:
+        logger.info("Using both Basic Auth and JWT Auth")
+        auth = basic_auth  # Basic Auth will be used for the request
+        headers.update(jwt_auth)  # Add JWT token as header
+    elif basic_auth:
+        logger.info("Using Basic Auth")
+        auth = basic_auth
+    elif jwt_auth:
+        logger.info("Using JWT Auth")
+        # Log the current JWT header name in DEBUG
+        logger.debug(f"Using JWT header name: {jwt_header_name}")
+        headers.update(jwt_auth)  # Only JWT token
+
     # If method is not provided use GET as default
     if method == "GET" or not method:
-        res = r.get("%s" % url, auth=auth, timeout=REQ_TIMEOUT, verify=REQ_TLS_VERIFY)
+        res = r.get("%s" % url, auth=auth, headers=headers, timeout=REQ_TIMEOUT, verify=REQ_TLS_VERIFY)
     elif method == "POST":
-        res = r.post("%s" % url, auth=auth, json=payload, timeout=REQ_TIMEOUT, verify=REQ_TLS_VERIFY)
+        res = r.post("%s" % url, auth=auth, headers=headers, json=payload, timeout=REQ_TIMEOUT, verify=REQ_TLS_VERIFY)
     else:
         logger.warning(f"Invalid REQ_METHOD: '{method}', please use 'GET' or 'POST'. Doing nothing.")
         return

src/requirements.txt

@@ -3,3 +3,4 @@ requests==2.32.3
 python-json-logger==3.2.1
 logfmter==0.0.9
 argparse==1.4.0
+pyjwt==2.10.1

test/resources/sidecar.yaml

@@ -427,3 +427,52 @@ type: Opaque
 stringData:
   username: "user1"
   password: "abcdefghijklmnopqrstuvwxyz"
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: sidecar-jwt-token
+  namespace: default
+spec:
+  serviceAccountName: sample-acc
+  containers:
+  - name: sidecar
+    image: kiwigrid/k8s-sidecar:testing
+    imagePullPolicy: IfNotPresent
+    volumeMounts:
+    - name: shared-volume
+      mountPath: /tmp/
+    - name: script-volume
+      mountPath: /opt/script.sh
+      subPath: script.sh
+    env:
+      - name: LABEL
+        value: "findme"
+      - name: FOLDER
+        value: /tmp/
+      - name: RESOURCE
+        value: both
+      - name: SCRIPT
+        value: "/opt/script.sh"
+      - name: JWT_HEADER_NAME
+        value: "some_jwt_header_name"
+      - name: LOG_LEVEL
+        value: "DEBUG"
+    envFrom:
+    - secretRef:
+        name: dummyyserver-jwt-token
+  volumes:
+  - name: shared-volume
+    emptyDir: {}
+  - name: script-volume
+    configMap:
+      name: script-configmap
+      defaultMode: 0777
+---  
+apiVersion: v1
+data:
+  JWT_TOKEN: ZXlKMGVYQWlPaUpLVjFRaUxDSmhiR2NpT2lKSVV6STFOaUo5LmV5SnBjM01pT2lKck9ITXRjMmxrWldOaGNpQjBaWE4wYVc1bklpd2lhV0YwSWpveE56TTROREV4TWpBd0xDSmxlSEFpT2pFM056QXdOelkzT1Rrc0ltRjFaQ0k2SW5kM2R5NXJhWGRwWjNKcFpDNWpiMjBpTENKemRXSWlPaUpLYjJodUlFUnZaU0lzSW5WelpYSnVZVzFsSWpvaVNtOW9ibTU1SWl3aVJXMWhhV3dpT2lKck9ITXRjMmxrWldOaGNrQnJhWGRwWjNKcFpDNWpiMjBpTENKU2IyeGxJam9pZEdWemRHVnlJbjAuRlFwYlBEb2lwdFVybDNWamY4a2lyZDg1VTY3RGR2c0FjZTVLazU1SFlDYw==
+kind: Secret
+metadata:
+  name: dummyyserver-jwt-token
+type: Opaque

test/server/server.py

@@ -37,12 +37,12 @@ async def read_item():
 
 
 @app.get("/secured", status_code=200, response_class=PlainTextResponse)
-async def read_secure_data(auth: HTTPBasicCredentials = Depends(basic_auth_scheme)):
-    if auth.username != 'user1' or auth.password != 'abcdefghijklmnopqrstuvwxyz':
-        logger.warning("[WARN] wrong auth: %s : %s ", auth.username, auth.password)
+async def read_secure_data(basic_auth: HTTPBasicCredentials = Depends(basic_auth_scheme)):
+    if basic_auth.username != 'user1' or basic_auth.password != 'abcdefghijklmnopqrstuvwxyz':
+        logger.warning("[WARN] wrong auth: %s : %s ", basic_auth.username, basic_auth.password)
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail=f"Incorrect user (${auth.username}) or password (${auth.password})",
+            detail=f"Incorrect user (${basic_auth.username}) or password (${basic_auth.password})",
             headers={"WWW-Authenticate": "Basic"},
         )
     return 'allowed'