chore(cloudrun): refactor to sample 'cloudrun_service_to_service_receive' and its test

run/service-auth/app.py

@@ -12,22 +12,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from http import HTTPStatus
 import os
 
 from flask import Flask, request
 
-from receive import receive_authorized_get_request
+from receive import receive_request_and_parse_auth_header
 
 app = Flask(__name__)
 
 
 @app.route("/")
-def main():
+def main() -> str:
     """Example route for receiving authorized requests."""
     try:
-        return receive_authorized_get_request(request)
+        response = receive_request_and_parse_auth_header(request)
+
+        status = HTTPStatus.UNAUTHORIZED
+        if "Hello" in response:
+            status = HTTPStatus.OK
+
+        return response, status
     except Exception as e:
-        return f"Error verifying ID token: {e}"
+        return f"Error verifying ID token: {e}", HTTPStatus.UNAUTHORIZED
 
 
 if __name__ == "__main__":

run/service-auth/receive.py

@@ -12,39 +12,49 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""
-Demonstrates how to receive authenticated service-to-service requests, eg
-for Cloud Run or Cloud Functions
+"""Demonstrates how to receive authenticated service-to-service requests.
+
+For example for Cloud Run or Cloud Functions.
 """
 
+# [START auth_validate_and_decode_bearer_token_on_flask]
 # [START cloudrun_service_to_service_receive]
+from flask import Request
 
+from google.auth.exceptions import GoogleAuthError
 from google.auth.transport import requests
 from google.oauth2 import id_token
 
 
-def receive_authorized_get_request(request):
-    """Parse the authorization header and decode the information
-    being sent by the Bearer token.
+def receive_request_and_parse_auth_header(request: Request) -> str:
+    """Parse the authorization header, validate the Bearer token
+    and decode the token to get its information.
 
     Args:
-        request: Flask request object
+        request: Flask request object.
 
     Returns:
-        The email from the request's Authorization header.
+        One of the following:
+        a) The email from the request's Authorization header.
+        b) A welcome message for anonymous users.
+        c) An error description.
     """
     auth_header = request.headers.get("Authorization")
     if auth_header:
-        # split the auth type and value from the header.
+        # Split the auth type and value from the header.
         auth_type, creds = auth_header.split(" ", 1)
 
         if auth_type.lower() == "bearer":
-            claims = id_token.verify_token(creds, requests.Request())
-            return f"Hello, {claims['email']}!\n"
-
+            # Find more information about `verify_token` function here:
+            # https://google-auth.readthedocs.io/en/master/reference/google.oauth2.id_token.html#google.oauth2.id_token.verify_token
+            try:
+                decoded_token = id_token.verify_token(creds, requests.Request())
+                return f"Hello, {decoded_token['email']}!\n"
+            except GoogleAuthError as e:
+                return f"Invalid token: {e}\n"
         else:
             return f"Unhandled header format ({auth_type}).\n"
-    return "Hello, anonymous user.\n"
-
 
+    return "Hello, anonymous user.\n"
 # [END cloudrun_service_to_service_receive]
+# [END auth_validate_and_decode_bearer_token_on_flask]

run/service-auth/receive_test.py

@@ -15,44 +15,80 @@
 # This test deploys a secure application running on Cloud Run
 # to test that the authentication sample works properly.
 
+from http import HTTPStatus
 import os
 import subprocess
 from urllib import error, request
 import uuid
 
 import pytest
+
 import requests
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
+from requests.sessions import Session
+
+PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
+
+STATUS_FORCELIST = [
+    HTTPStatus.BAD_REQUEST,
+    HTTPStatus.UNAUTHORIZED,
+    HTTPStatus.FORBIDDEN,
+    HTTPStatus.NOT_FOUND,
+    HTTPStatus.INTERNAL_SERVER_ERROR,
+    HTTPStatus.BAD_GATEWAY,
+    HTTPStatus.SERVICE_UNAVAILABLE,
+    HTTPStatus.GATEWAY_TIMEOUT,
+],
 
 
-@pytest.fixture()
-def services():
-    # Unique suffix to create distinct service names
-    suffix = uuid.uuid4().hex
-    service_name = f"receive-{suffix}"
-    project = os.environ["GOOGLE_CLOUD_PROJECT"]
+@pytest.fixture(scope="module")
+def service_name() -> str:
+    # Add a unique suffix to create distinct service names.
+    service_name_str = f"receive-{uuid.uuid4().hex}"
 
-    # Deploy receive Cloud Run Service
+    # Deploy the Cloud Run Service.
     subprocess.run(
         [
             "gcloud",
             "run",
             "deploy",
-            service_name,
+            service_name_str,
             "--project",
-            project,
+            PROJECT_ID,
             "--source",
             ".",
             "--region=us-central1",
             "--allow-unauthenticated",
             "--quiet",
         ],
+        # Rise a CalledProcessError exception for a non-zero exit code.
         check=True,
     )
 
-    # Get the URL for the service
-    endpoint_url = (
+    yield service_name_str
+
+    # Clean-up after running the test.
+    subprocess.run(
+        [
+            "gcloud",
+            "run",
+            "services",
+            "delete",
+            service_name_str,
+            "--project",
+            PROJECT_ID,
+            "--async",
+            "--region=us-central1",
+            "--quiet",
+        ],
+        check=True,
+    )
+
+
+@pytest.fixture(scope="module")
+def endpoint_url(service_name: str) -> str:
+    endpoint_url_str = (
         subprocess.run(
             [
                 "gcloud",
@@ -61,7 +97,7 @@ def services():
                 "describe",
                 service_name,
                 "--project",
-                project,
+                PROJECT_ID,
                 "--region=us-central1",
                 "--format=value(status.url)",
             ],
@@ -72,7 +108,12 @@ def services():
         .decode()
     )
 
-    token = (
+    return endpoint_url_str
+
+
+@pytest.fixture(scope="module")
+def token() -> str:
+    token_str = (
         subprocess.run(
             ["gcloud", "auth", "print-identity-token"],
             stdout=subprocess.PIPE,
@@ -82,38 +123,20 @@ def services():
         .decode()
     )
 
-    yield endpoint_url, token
-
-    subprocess.run(
-        [
-            "gcloud",
-            "run",
-            "services",
-            "delete",
-            service_name,
-            "--project",
-            project,
-            "--async",
-            "--region=us-central1",
-            "--quiet",
-        ],
-        check=True,
-    )
-
+    return token_str
 
-def test_auth(services):
-    url = services[0]
-    token = services[1]
 
-    req = request.Request(url)
+@pytest.fixture(scope="module")
+def client(endpoint_url: str) -> Session:
+    req = request.Request(endpoint_url)
     try:
         _ = request.urlopen(req)
     except error.HTTPError as e:
-        assert e.code == 403
+        assert e.code == HTTPStatus.FORBIDDEN
 
     retry_strategy = Retry(
         total=3,
-        status_forcelist=[400, 401, 403, 404, 500, 502, 503, 504],
+        status_forcelist=STATUS_FORCELIST,
         allowed_methods=["GET", "POST"],
         backoff_factor=3,
     )
@@ -122,8 +145,34 @@ def test_auth(services):
     client = requests.session()
     client.mount("https://", adapter)
 
-    response = client.get(url, headers={"Authorization": f"Bearer {token}"})
+    return client
+
+
+def test_authentication_on_cloud_run(
+    client: Session, endpoint_url: str, token: str
+) -> None:
+    response = client.get(
+        endpoint_url, headers={"Authorization": f"Bearer {token}"}
+    )
+    response_content = response.content.decode("utf-8")
+
+    assert response.status_code == HTTPStatus.OK
+    assert "Hello" in response_content
+    assert "anonymous" not in response_content
+
+
+def test_anonymous_request_on_cloud_run(client: Session, endpoint_url: str) -> None:
+    response = client.get(endpoint_url)
+    response_content = response.content.decode("utf-8")
+
+    assert response.status_code == HTTPStatus.OK
+    assert "Hello" in response_content
+    assert "anonymous" in response_content
+
+
+def test_invalid_token(client: Session, endpoint_url: str) -> None:
+    response = client.get(
+        endpoint_url, headers={"Authorization": "Bearer i-am-not-a-real-token"}
+    )
 
-    assert response.status_code == 200
-    assert "Hello" in response.content.decode("UTF-8")
-    assert "anonymous" not in response.content.decode("UTF-8")
+    assert response.status_code == HTTPStatus.UNAUTHORIZED

run/service-auth/requirements-test.txt

@@ -1 +1 @@
-pytest==8.2.0
+pytest==8.3.5

run/service-auth/requirements.txt

@@ -1,5 +1,5 @@
 google-auth==2.38.0
-requests==2.31.0
-Flask==3.0.3
+requests==2.32.3
+Flask==3.1.0
 gunicorn==23.0.0
-Werkzeug==3.0.3
+Werkzeug==3.1.3