def webhook_ingress_internal_only_status()

in dialogflow-cx/vpc-sc-demo/backend/status_blueprint.py [0:0]

• 31 lines of code
• 5 McCabe index (conditional complexity)

def webhook_ingress_internal_only_status():
    """Get boolean status of internally restricted webhook ingress."""
    data = su.get_token_and_project(flask.request)
    if "response" in data:
        return data["response"]
    project_id, token = data["project_id"], data["token"]
    region = flask.request.args["region"]
    webhook_name = flask.request.args["webhook_name"]

    response = su.check_function_exists(token, project_id, region, webhook_name)
    if "response" in response:
        return response["response"]

    headers = {}
    headers["x-goog-user-project"] = project_id
    headers["Authorization"] = f"Bearer {token}"
    result = requests.get(
        (
            "https://cloudfunctions.googleapis.com/v1/projects/"
            f"{project_id}/locations/{region}/functions/{webhook_name}"
        ),
        headers=headers,
        timeout=10,
    )
    if result.status_code != 200:
        logger.info("  cloudfunctions API rejected request: %s", result.text)
        return flask.abort(result.status_code)
    result_dict = result.json()
    if result_dict["ingressSettings"] in [
        "ALLOW_INTERNAL_ONLY",
        "ALLOW_INTERNAL_AND_GCLB",
    ]:
        return flask.Response(status=200, response=json.dumps({"status": True}))
    return flask.Response(status=200, response=json.dumps({"status": False}))