fix(cloudrun): fix 'cloudrun_service_to_service_receive' sample

run/service-auth/app.py

@@ -12,30 +12,133 @@
 # 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 urllib.request import Request, urlopen
 
 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.cloud import run_v2
+from google.oauth2 import id_token
+
+# Get the Service Name as found in Cloud Run.
+SERVICE_NAME = os.getenv("K_SERVICE")
+
+# Get the Project ID.
+req = Request("http://metadata.google.internal/computeMetadata/v1/project/project-id")
+req.add_header("Metadata-Flavor", "Google")
+PROJECT_ID = urlopen(req).read().decode("utf-8")
+
+# Get the Region.
+req = Request("http://metadata.google.internal/computeMetadata/v1/instance/region")
+req.add_header("Metadata-Flavor", "Google")
+
+# Returns "projects/PROJECT-NUMBER/regions/REGION"
+project_region_list = urlopen(req).read().decode("utf-8").split('/')
+
+REGION = project_region_list[3]
+
+# Get the Service URL, required to define the valid audience for the Token.
+# https://cloud.google.com/run/docs/triggering/https-request#deterministic
+FULL_SERVICE_NAME = f"projects/{PROJECT_ID}/locations/{REGION}/services/{SERVICE_NAME}"
+
+client = run_v2.ServicesClient()
+
+service_request = run_v2.GetServiceRequest(
+    name=FULL_SERVICE_NAME,
+)
+
+SERVICE_URI = client.get_service(request=service_request).uri
 
 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
+
+    # The token audience will be the SERVICE_URL.
+    # If `audience` was None, it won't be verified.
+    audience = SERVICE_URI
+
+    if auth_type.lower() == "bearer":
+        # Get the ID token.
+        # Find more info about the ID Token here:
+        # https://cloud.google.com/docs/authentication/token-types#id
+
+        try:
+            # Find more information about `verify_oauth2_token` function here:
+            # 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,
+            )
+
+            # Verify that the token contains the email claims.
+            if decoded_token['email_verified']:
+                print(f"Email verified {decoded_token['email']}")
+
+                return decoded_token['email']
+
+            print("Email wasn't verified.")
+            return None
+        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 → 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
@@ -45,7 +49,7 @@
 @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}"
+    service_name_str = f"receive-python-{uuid.uuid4().hex}"
 
     # Deploy the Cloud Run Service.
     subprocess.run(
@@ -112,28 +116,21 @@ def endpoint_url(service_name: str) -> str:
 
 
 @pytest.fixture(scope="module")
-def token() -> str:
-    token_str = (
-        subprocess.run(
-            ["gcloud", "auth", "print-identity-token"],
-            stdout=subprocess.PIPE,
-            check=True,
-        )
-        .stdout.strip()
-        .decode()
-    )
+def token(endpoint_url: str) -> str:
+    # Cloud Run uses your service's hostname as the `audience` value.
+    # For example: 'https://my-cloud-run-service.run.app'
+    target_audience = endpoint_url
+    auth_req = transport_requests.Request()
 
-    return token_str
+    # More info for the `fetch_id_token`function
+    # https://googleapis.dev/python/google-auth/1.14.0/reference/google.oauth2.id_token.html
+    token = id_token.fetch_id_token(auth_req, target_audience)
 
+    return token
 
-@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
 
+@pytest.fixture(scope="module")
+def client() -> Session:
     retry_strategy = Retry(
         total=3,
         status_forcelist=STATUS_FORCELIST,
@@ -148,7 +145,8 @@ def client(endpoint_url: str) -> Session:
     return client
 
 
-def test_authentication_on_cloud_run(
+@backoff.on_exception(backoff.expo, Exception, max_time=60)
+def test_authentication_on_cloud_run_service(
     client: Session, endpoint_url: str, token: str
 ) -> None:
     response = client.get(
@@ -158,19 +156,15 @@ def test_authentication_on_cloud_run(
 
     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:
+def test_anonymous_request_on_cloud_run_service(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
+    assert response.status_code == HTTPStatus.UNAUTHORIZED
 
 
-def test_invalid_token(client: Session, endpoint_url: str) -> None:
+def test_an_invalid_token_on_cloud_run_service(client: Session, endpoint_url: str) -> None:
     response = client.get(
         endpoint_url, headers={"Authorization": "Bearer i-am-not-a-real-token"}
     )

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.0
 gunicorn==23.0.0
-Werkzeug==3.1.3