33import json
44import os
55import subprocess
6- import time
7- from datetime import datetime , timedelta
86from typing import Generator
97
108import httpx
1917
2018@pytest .fixture (scope = "module" )
2119def mock_oidc_server () -> Generator [str , None , None ]:
22- """Use or deploy mock OIDC server for testing."""
20+ """Use Helm-deployed mock OIDC server for testing."""
2321 namespace = os .getenv ("NAMESPACE" , "eoapi" )
2422 release_name = os .getenv ("RELEASE_NAME" , "eoapi" )
25- deployed_by_fixture = False
2623
27- # Check if mock OIDC server is already deployed by helm
24+ # Check if mock OIDC server is deployed by Helm
2825 result = subprocess .run (
2926 [
3027 "kubectl" ,
@@ -40,201 +37,39 @@ def mock_oidc_server() -> Generator[str, None, None]:
4037 text = True ,
4138 )
4239
43- if result .returncode == 0 :
44- # Mock OIDC server is already deployed by helm
45- deployment = json .loads (result .stdout )
46- if deployment .get ("status" , {}).get ("readyReplicas" , 0 ) > 0 :
47- # Server is ready, use it
48- oidc_url = f"http://{ release_name } { namespace }
49- yield oidc_url
50- return # Don't cleanup helm-deployed server
51-
52- # If not deployed by helm, deploy it manually for testing
53- deployed_by_fixture = True
54- deployment_yaml = f"""
55- apiVersion: apps/v1
56- kind: Deployment
57- metadata:
58- name: mock-oidc-server
59- namespace: { namespace }
60- spec:
61- replicas: 1
62- selector:
63- matchLabels:
64- app: mock-oidc-server
65- template:
66- metadata:
67- labels:
68- app: mock-oidc-server
69- spec:
70- containers:
71- - name: mock-oidc
72- image: ghcr.io/alukach/mock-oidc-server:latest
73- env:
74- - name: MOCK_OIDC_PORT
75- value: "8888"
76- - name: MOCK_OIDC_CLIENT_ID
77- value: "test-client"
78- - name: MOCK_OIDC_CLIENT_SECRET
79- value: "test-secret"
80- ports:
81- - containerPort: 8888
82- ---
83- apiVersion: v1
84- kind: Service
85- metadata:
86- name: mock-oidc-server
87- namespace: { namespace }
88- spec:
89- selector:
90- app: mock-oidc-server
91- ports:
92- - port: 8080
93- targetPort: 8888
94- """
95-
96- # Apply deployment
97- result = subprocess .run (
98- ["kubectl" , "apply" , "-f" , "-" ],
99- input = deployment_yaml ,
100- text = True ,
101- capture_output = True ,
102- )
10340 if result .returncode != 0 :
104- pytest .skip (f"Failed to deploy mock OIDC server: { result .stderr } )
105-
106- # Wait for pod to be ready
107- for _ in range (30 ):
108- result = subprocess .run (
109- [
110- "kubectl" ,
111- "get" ,
112- "pods" ,
113- "-n" ,
114- namespace ,
115- "-l" ,
116- "app=mock-oidc-server" ,
117- "--field-selector=status.phase=Running" ,
118- "-o" ,
119- "json" ,
120- ],
121- capture_output = True ,
122- text = True ,
123- )
124- if result .returncode == 0 :
125- pods = json .loads (result .stdout )
126- if pods .get ("items" ):
127- break
128- time .sleep (2 )
129- else :
130- pytest .skip ("Mock OIDC server failed to start" )
131-
132- # Update stac-auth-proxy to use mock OIDC server (only if deployed by fixture)
133- oidc_url = f"http://mock-oidc-server.{ namespace }
134- if deployed_by_fixture :
135- subprocess .run (
136- [
137- "kubectl" ,
138- "set" ,
139- "env" ,
140- f"deployment/{ release_name } ,
141- f"OIDC_DISCOVERY_URL={ oidc_url } ,
142- "DEFAULT_PUBLIC=true" ,
143- "-n" ,
144- namespace ,
145- ],
146- capture_output = True ,
41+ pytest .skip (
42+ "Mock OIDC server not deployed. Enable mockOidcServer in Helm values."
14743 )
14844
149- # Wait for stac-auth-proxy to restart and become ready
150- time .sleep (5 )
151- for _ in range (30 ):
152- result = subprocess .run (
153- [
154- "kubectl" ,
155- "get" ,
156- "pods" ,
157- "-n" ,
158- namespace ,
159- "-l" ,
160- f"app.kubernetes.io/instance={ release_name } ,
161- "--field-selector=status.phase=Running" ,
162- "-o" ,
163- "json" ,
164- ],
165- capture_output = True ,
166- text = True ,
167- )
168- if result .returncode == 0 :
169- pods = json .loads (result .stdout )
170- if pods .get ("items" ) and len (pods ["items" ]) > 0 :
171- # Check if auth proxy started successfully with new config
172- result = subprocess .run (
173- [
174- "kubectl" ,
175- "logs" ,
176- f"deployment/{ release_name } ,
177- "-n" ,
178- namespace ,
179- "--tail=5" ,
180- ],
181- capture_output = True ,
182- text = True ,
183- )
184- if "Application startup complete" in result .stdout :
185- break
186- time .sleep (2 )
45+ # Verify the server is ready
46+ deployment = json .loads (result .stdout )
47+ if deployment .get ("status" , {}).get ("readyReplicas" , 0 ) == 0 :
48+ pytest .skip ("Mock OIDC server not ready" )
18749
50+ # Return the service URL - no manual deployment or cleanup needed
51+ oidc_url = f"http://{ release_name } { namespace }
18852 yield oidc_url
18953
190- # Cleanup only if deployed by this fixture
191- if deployed_by_fixture :
192- subprocess .run (
193- [
194- "kubectl" ,
195- "delete" ,
196- "deployment" ,
197- "mock-oidc-server" ,
198- "-n" ,
199- namespace ,
200- ],
201- capture_output = True ,
202- )
203- subprocess .run (
204- [
205- "kubectl" ,
206- "delete" ,
207- "service" ,
208- "mock-oidc-server" ,
209- "-n" ,
210- namespace ,
211- ],
212- capture_output = True ,
213- )
214-
21554
21655@pytest .fixture
217- def valid_token (mock_oidc_server : str ) -> str :
218- """Generate valid JWT token."""
219- try :
220- import jwt
221- except ImportError :
222- pytest .skip ("pyjwt not installed" )
56+ def dummy_token (mock_oidc_server : str ) -> str :
57+ """Create a dummy JWT token for testing auth proxy behavior.
58+
59+ This is a dummy token used to test that the auth proxy is correctly
60+ processing Authorization headers and rejecting invalid tokens.
61+
62+ Args:
63+ mock_oidc_server: URL of the mock OIDC server (unused but required for fixture)
22364
224- payload = {
225- "sub" : "test-user" ,
226- 227- "iss" : mock_oidc_server ,
228- "aud" : "test-client" ,
229- "exp" : datetime .utcnow () + timedelta (hours = 1 ),
230- "iat" : datetime .utcnow (),
231- }
232- # Mock OIDC server uses a simple secret
233- token = jwt .encode (payload , "test-secret" , algorithm = "HS256" )
234- # Handle both str (newer PyJWT) and bytes (older PyJWT) return types
235- if isinstance (token , bytes ):
236- return token .decode ("utf-8" )
237- return str (token )
65+ Returns:
66+ A dummy Bearer token string for testing
67+
68+ Note:
69+ The actual token validation will fail, which is the expected behavior
70+ since creating valid OIDC tokens requires complex authorization flows.
71+ """
72+ return "Bearer dummy-token-for-testing"
23873
23974
24075def test_stac_write_create_item_without_token (
@@ -271,3 +106,77 @@ def test_stac_write_create_item_without_token(
271106 assert resp .status_code in [401 , 403 ], (
272107 f"Expected 401 Unauthorized or 403 Forbidden, got { resp .status_code }
273108 )
109+
110+
111+ def test_stac_write_create_item_with_dummy_token (
112+ stac_endpoint : str , mock_oidc_server : str , dummy_token : str
113+ ) -> None :
114+ """Test that auth proxy correctly processes and rejects invalid tokens.
115+
116+ This test verifies the auth proxy is working by sending a request with
117+ an invalid Authorization header. The expected behavior is rejection
118+ with 401 or 403 status code.
119+
120+ Args:
121+ stac_endpoint: The STAC API endpoint URL
122+ mock_oidc_server: URL of the mock OIDC server (for test context)
123+ dummy_token: A dummy Bearer token for testing
124+
125+ Validates:
126+ - Auth proxy intercepts requests with Authorization headers
127+ - Invalid tokens are properly rejected (401/403 response)
128+ - Auth system is functioning correctly
129+ """
130+ # Attempt to create an item with invalid authorization
131+ headers = {"Authorization" : dummy_token }
132+ resp = client .post (
133+ f"{ stac_endpoint } ,
134+ headers = headers ,
135+ json = {
136+ "id" : "test-auth-item-with-dummy-token" ,
137+ "type" : "Feature" ,
138+ "stac_version" : "1.0.0" ,
139+ "properties" : {"datetime" : "2024-01-01T00:00:00Z" },
140+ "geometry" : {"type" : "Point" , "coordinates" : [0 , 0 ]},
141+ "links" : [],
142+ "assets" : {},
143+ "collection" : "noaa-emergency-response" ,
144+ "bbox" : [- 0.1 , - 0.1 , 0.1 , 0.1 ],
145+ },
146+ )
147+
148+ # Verify auth proxy correctly rejects invalid token
149+ assert resp .status_code in [401 , 403 ], (
150+ f"Expected 401 Unauthorized or 403 Forbidden with invalid token, "
151+ f"got { resp .status_code }
152+ f"properly validating Authorization headers."
153+ )
154+
155+
156+ def test_auth_proxy_integration_with_mock_oidc () -> None :
157+ """Integration test to verify auth proxy can communicate with mock OIDC server.
158+
159+ This test verifies that the auth proxy can successfully:
160+ 1. Connect to the mock OIDC server
161+ 2. Retrieve OIDC configuration
162+ 3. Access JWKS for token validation
163+ 4. Validate properly signed JWT tokens
164+ 5. Allow authenticated write operations
165+
166+ Implementation approaches:
167+ - Run tests in-cluster (e.g., as a Job or from another pod)
168+ - Use kubectl port-forward to expose mock OIDC server
169+ - Extract and use the mock server's actual RSA private key
170+ - Create tokens using the server's token generation endpoint
171+
172+ Note: This test is currently skipped because it requires network access to
173+ internal k8s services that aren't available from external test environments.
174+ """
175+ pytest .skip (
176+ "This test requires network access to internal k8s services. "
177+ "To implement:\n "
178+ "1. Set up kubectl port-forward for mock OIDC server\n "
179+ "2. Get valid token from http://localhost:PORT/ endpoint\n "
180+ "3. Test STAC write operations with valid Authorization header\n "
181+ "4. Verify 200/201 responses for authenticated requests"
182+ )
0 commit comments