fix(cloudrun): fix 'cloudrun_service_to_service_receive' sample (#13372)

* fix(cloudrun): Proof of Concept - Service to Service.

Implement a Flask app which receives ID tokens, and returns an HTTP response based on the validation of the token.

* fix(cloudrun): Proof of Concept - fix sample to pass Unit Tests

* fix(cloudrun): fixes to pass unit tests

* fix(cloudrun): delete draft

* fix(cloudrun): add feedback from code-assist

- Add a try-except block to handle potential ValueError.

* fix(cloudrun): delete placeholder variable

* fix(cloudrun): fix typo

* fix(cloudrun): PoC Replace gcloud commands with a Client Library

* fix(cloudrun): clean up before sharing for internal review

* fix(cloudrun): delete proof of concept to replace gcloud commands with a client library

* Revert "fix(cloudrun): delete proof of concept to replace gcloud commands with a client library"

This reverts commit 4ddb193.

* Revert "fix(cloudrun): clean up before sharing for internal review"

This reverts commit a8486ba.

* Revert "fix(cloudrun): PoC Replace gcloud commands with a Client Library"

This reverts commit 48f54a7.

* fix(cloudrun): rename 'test_invalid_token'

PR Review #13372 (comment)

* fix(cloudrun): apply feedback from PR Review

#13372 (review)

- Change from getting the service URL from the Metadata server to an env var supplied in the Deployment
- Fix comments
- Change from getting the URL from the `gcloud run services descrive` to the same Client library used in app.py to unify the URL
- Rename tests

* fix(cloudrun): remove duplicated code to get the Service URL

* fix(cloudrun): code cleanup before review

* fix(cloudrun): implement reading the Service URL from an env var

#13372 (comment)

* fix(cloudrun): directly return the id token in the fixture

* fix(cloudrun): delete unused import

Author: eapl-gemugami

run/service-auth/app.py

@@ -12,30 +12,104 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# [START auth_validate_and_decode_bearer_token_on_flask]
+# [START cloudrun_service_to_service_receive]
+"""Demonstrates how to receive authenticated service-to-service requests
+on a Cloud Run Service.
+"""
+
 from http import HTTPStatus
 import os
+from typing import Optional
 
 from flask import Flask, request
 
-from receive import receive_request_and_parse_auth_header
+from google.auth.exceptions import GoogleAuthError
+from google.auth.transport import requests
+from google.oauth2 import id_token
 
 app = Flask(__name__)
 
 
+def parse_auth_header(auth_header: str) -> Optional[str]:
+    """Parse the authorization header, validate and decode the Bearer token.
+
+    Args:
+        auth_header: Raw HTTP header with a Bearer token.
+
+    Returns:
+        A string containing the email from the token.
+        None if the token is invalid or the email can't be retrieved.
+    """
+
+    # Split the auth type and value from the header.
+    try:
+        auth_type, creds = auth_header.split(" ", 1)
+    except ValueError:
+        print("Malformed Authorization header.")
+        return None
+
+    # Get the service URL from the environment variable
+    # set at the time of deployment.
+    service_url = os.environ["SERVICE_URL"]
+
+    # Define the expected audience as the Service Base URL.
+    audience = service_url
+
+    # Validate and decode the ID token in the header.
+    if auth_type.lower() == "bearer":
+        try:
+            # Find more information about `verify_oauth2_token` function:
+            # https://googleapis.dev/python/google-auth/latest/reference/google.oauth2.id_token.html#google.oauth2.id_token.verify_oauth2_token
+            decoded_token = id_token.verify_oauth2_token(
+                id_token=creds,
+                request=requests.Request(),
+                audience=audience,
+            )
+
+            # More info about the structure for the decoded ID Token here:
+            # https://cloud.google.com/docs/authentication/token-types#id
+
+            # Verify that the token contains the email claim.
+            if decoded_token['email_verified']:
+                print(f"Email verified: {decoded_token['email']}")
+
+                return decoded_token['email']
+
+            print("Invalid token. Email wasn't verified.")
+        except GoogleAuthError as e:
+            print(f"Invalid token: {e}")
+    else:
+        print(f"Unhandled header format ({auth_type}).")
+
+    return None
+
+
 @app.route("/")
 def main() -> str:
-    """Example route for receiving authorized requests."""
+    """Example route for receiving authorized requests only."""
     try:
-        response = receive_request_and_parse_auth_header(request)
+        auth_header = request.headers.get("Authorization")
+        if auth_header:
+            email = parse_auth_header(auth_header)
+
+            if email:
+                return f"Hello, {email}.\n", HTTPStatus.OK
 
-        status = HTTPStatus.UNAUTHORIZED
-        if "Hello" in response:
-            status = HTTPStatus.OK
+        # Indicate that the request must be authenticated
+        # and that Bearer auth is the permitted authentication scheme.
+        headers = {"WWW-Authenticate": "Bearer"}
 
-        return response, status
+        return (
+            "Unauthorized request. Please supply a valid bearer token.",
+            HTTPStatus.UNAUTHORIZED,
+            headers,
+        )
     except Exception as e:
         return f"Error verifying ID token: {e}", HTTPStatus.UNAUTHORIZED
 
 
 if __name__ == "__main__":
     app.run(host="localhost", port=int(os.environ.get("PORT", 8080)), debug=True)
+# [END cloudrun_service_to_service_receive]
+# [END auth_validate_and_decode_bearer_token_on_flask]

run/service-auth/receive.py

@@ -17,6 +17,8 @@
 For example for Cloud Run or Cloud Functions.
 """
 
+# This sample will be migrated to app.py
+
 # [START auth_validate_and_decode_bearer_token_on_flask]
 # [START cloudrun_service_to_service_receive]
 from flask import Request

run/service-auth/receive_test.py renamed to run/service-auth/receive_auth_requests_test.py

@@ -13,14 +13,18 @@
 # limitations under the License.
 
 # This test deploys a secure application running on Cloud Run
-# to test that the authentication sample works properly.
+# to validate receiving authenticated requests.
 
 from http import HTTPStatus
 import os
 import subprocess
-from urllib import error, request
 import uuid
 
+import backoff
+
+from google.auth.transport import requests as transport_requests
+from google.oauth2 import id_token
+
 import pytest
 
 import requests
@@ -29,6 +33,7 @@
 from requests.sessions import Session
 
 PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
+REGION = "us-central1"
 
 STATUS_FORCELIST = [
     HTTPStatus.BAD_REQUEST,
@@ -43,30 +48,55 @@
 
 
 @pytest.fixture(scope="module")
-def service_name() -> str:
+def project_number() -> str:
+    return (
+        subprocess.run(
+            [
+                "gcloud",
+                "projects",
+                "describe",
+                PROJECT_ID,
+                "--format=value(projectNumber)",
+            ],
+            stdout=subprocess.PIPE,
+            check=True,
+        )
+        .stdout.strip()
+        .decode()
+    )
+
+
+@pytest.fixture(scope="module")
+def service_url(project_number: str) -> str:
+    """Deploys a Run Service and returns its Base URL."""
+
     # Add a unique suffix to create distinct service names.
-    service_name_str = f"receive-{uuid.uuid4().hex}"
+    service_name = f"receive-python-{uuid.uuid4().hex}"
 
-    # Deploy the Cloud Run Service.
+    # Construct the Deterministic URL.
+    service_url = f"https://{service_name}-{project_number}.{REGION}.run.app"
+
+    # Deploy the Cloud Run Service supplying the URL as an environment variable.
     subprocess.run(
         [
             "gcloud",
             "run",
             "deploy",
-            service_name_str,
+            service_name,
             "--project",
             PROJECT_ID,
             "--source",
             ".",
-            "--region=us-central1",
+            f"--region={REGION}",
             "--allow-unauthenticated",
+            f"--set-env-vars=SERVICE_URL={service_url}",
             "--quiet",
         ],
         # Rise a CalledProcessError exception for a non-zero exit code.
         check=True,
     )
 
-    yield service_name_str
+    yield service_url
 
     # Clean-up after running the test.
     subprocess.run(
@@ -75,65 +105,27 @@ def service_name() -> str:
             "run",
             "services",
             "delete",
-            service_name_str,
+            service_name,
             "--project",
             PROJECT_ID,
             "--async",
-            "--region=us-central1",
+            f"--region={REGION}",
             "--quiet",
         ],
         check=True,
     )
 
 
 @pytest.fixture(scope="module")
-def endpoint_url(service_name: str) -> str:
-    endpoint_url_str = (
-        subprocess.run(
-            [
-                "gcloud",
-                "run",
-                "services",
-                "describe",
-                service_name,
-                "--project",
-                PROJECT_ID,
-                "--region=us-central1",
-                "--format=value(status.url)",
-            ],
-            stdout=subprocess.PIPE,
-            check=True,
-        )
-        .stdout.strip()
-        .decode()
-    )
+def token(service_url: str) -> str:
+    auth_req = transport_requests.Request()
+    target_audience = service_url
 
-    return endpoint_url_str
+    return id_token.fetch_id_token(auth_req, target_audience)
 
 
 @pytest.fixture(scope="module")
-def token() -> str:
-    token_str = (
-        subprocess.run(
-            ["gcloud", "auth", "print-identity-token"],
-            stdout=subprocess.PIPE,
-            check=True,
-        )
-        .stdout.strip()
-        .decode()
-    )
-
-    return token_str
-
-
-@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 == HTTPStatus.FORBIDDEN
-
+def client() -> Session:
     retry_strategy = Retry(
         total=3,
         status_forcelist=STATUS_FORCELIST,
@@ -148,31 +140,28 @@ def client(endpoint_url: str) -> Session:
     return client
 
 
-def test_authentication_on_cloud_run(
-    client: Session, endpoint_url: str, token: str
+@backoff.on_exception(backoff.expo, Exception, max_time=60)
+def test_authenticated_request(
+    client: Session, service_url: str, token: str,
 ) -> None:
     response = client.get(
-        endpoint_url, headers={"Authorization": f"Bearer {token}"}
+        service_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")
+def test_anonymous_request(client: Session, service_url: str) -> None:
+    response = client.get(service_url)
 
-    assert response.status_code == HTTPStatus.OK
-    assert "Hello" in response_content
-    assert "anonymous" in response_content
+    assert response.status_code == HTTPStatus.UNAUTHORIZED
 
 
-def test_invalid_token(client: Session, endpoint_url: str) -> None:
+def test_invalid_token(client: Session, service_url: str) -> None:
     response = client.get(
-        endpoint_url, headers={"Authorization": "Bearer i-am-not-a-real-token"}
+        service_url, headers={"Authorization": "Bearer i-am-not-a-real-token"}
     )
 
     assert response.status_code == HTTPStatus.UNAUTHORIZED

run/service-auth/requirements-test.txt

@@ -1 +1,2 @@
 pytest==8.3.5
+backoff==2.2.1

run/service-auth/requirements.txt

@@ -1,5 +1,5 @@
-google-auth==2.38.0
+google-auth==2.40.1
+google-cloud-run==0.10.18
 requests==2.32.3
 Flask==3.1.1
 gunicorn==23.0.0
-Werkzeug==3.1.3