Health Endpoint Validation

README.md

@@ -57,6 +57,10 @@ This sample uses FastAPI as a baseline which is a scalable, modern, fast and pro
 
 This sample uses Open Telemetry and the FastAPI and Azure Application Insights integration for end to end tracking of API calls using logs and metrics and dependency calls. More information about Open Telemetry can be found [here](https://opentelemetry.io/).
 
+### Health Endpoint
+
+This sample exposes a health endpoint that includes header validation (Header `x-ms-auth-internal-token`) to ensure that only the health check feature of the Azure Function it self is allowed to call this endpoint. When trying to reach this endpoint from outside, the request will be blocked with a 400 response.
+
 ### Testing
 
 Testing of the Azure Functon application code. The testing is done using `pytest`. Tests are stored in the [`/tests` folder](/tests/) and should be extended for new functionality that is being implemented over time. The `pytest.ini` is used to reference the Azure Functon project for imports. This file makes sure that the respective python objects from the Azrue Function application code can be imported into the tests and validated accordingly.

code/function/fastapp/api/v1/endpoints/heartbeat.py

@@ -1,6 +1,7 @@
 from typing import Any
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+from fastapp.health.validate_request import verify_health_auth_header
 from fastapp.models.heartbeat import HearbeatResult
 from fastapp.utils import setup_logging
 
@@ -9,7 +10,12 @@
 router = APIRouter()
 
 
-@router.get("/heartbeat", response_model=HearbeatResult, name="heartbeat")
+@router.get(
+    "/heartbeat",
+    response_model=HearbeatResult,
+    name="heartbeat",
+    dependencies=[Depends(verify_health_auth_header)],
+)
 async def get_hearbeat() -> Any:
     logger.info("Received Heartbeat Request")
     return HearbeatResult(isAlive=True)

code/function/fastapp/core/config.py

@@ -18,6 +18,9 @@ class Settings(BaseSettings):
     )
     WEBSITE_NAME: str = Field(default="test", alias="WEBSITE_SITE_NAME")
     WEBSITE_INSTANCE_ID: str = Field(default="0", alias="WEBSITE_INSTANCE_ID")
+    WEBSITE_AUTH_ENCRYPTION_KEY: str = Field(
+        default="", alias="WEBSITE_AUTH_ENCRYPTION_KEY"
+    )
     MY_SECRET_CONFIG: str = Field(default="", alias="MY_SECRET_CONFIG")
 
 

code/function/fastapp/health/validate_request.py

@@ -0,0 +1,27 @@
+import base64
+from hashlib import sha256
+from typing import Annotated
+
+from fastapi import Header, HTTPException
+from fastapp.core.config import settings
+
+
+async def verify_health_auth_header(
+    x_ms_auth_internal_token: Annotated[str, Header()]
+) -> bool:
+    """Returns true if SHA256 of header_value matches WEBSITE_AUTH_ENCRYPTION_KEY.
+    Documentation: https://learn.microsoft.com/en-us/azure/app-service/monitor-instances-health-check?tabs=python#authentication-and-security
+
+    x_ms_auth_internal_token: Value of the x-ms-auth-internal-token header.
+    RETURNS (bool): Specifies whether the header matches.
+    """
+    website_auth_encryption_key = settings.WEBSITE_AUTH_ENCRYPTION_KEY
+    hash = base64.b64encode(
+        sha256(website_auth_encryption_key.encode("utf-8")).digest()
+    ).decode("utf-8")
+    if hash != x_ms_auth_internal_token:
+        raise HTTPException(
+            status_code=400, detail="x-ms-auth-internal-token is invalid"
+        )
+    else:
+        return True

tests/test_main.py

@@ -12,9 +12,12 @@ def client() -> TestClient:
 def test_get_heartbeat(client, version):
     # arrange
     path = f"/{version}/health/heartbeat"
+    headers = {
+        "x-ms-auth-internal-token": "47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU="
+    }
 
     # action
-    response = client.get(path)
+    response = client.get(path, headers=headers)
 
     # assert
     assert response.status_code == 200