Developing Starlark extensions

def envoy_check(req):
    path = req.attributes.request.http.path
    # Deny the request if the path starts with /restricted
    if path.startswith("/restricted/"):
        return struct(envoy_check_response = struct(
            status = struct(code = 7),
            denied_response = struct(body = "Go away")
        ))

    # Allow the request to go through with one of the headers removed
    return struct(envoy_check_response = struct(
        status = struct(code = 0),
        ok_response = struct(
            headers_to_remove = ["x-confidential-header"]
        )
    ))

def envoy_check(req):
    http_req = req.attributes.request.http
    path = http_req.path

    if path in context.extension_config["secure_paths"]:
        return struct(cerbos_mapping = struct(
            check_request = struct(
                principal = struct(id = "daffy", roles = ["user"]),
                resources = [struct(
                    resource = struct(id = "x", kind = "request", attr = {"path": path}),
                    actions = [http_req.method]
                )]
            ),
            allow_response = struct(
                status = struct(code = 0),
                ok_response = struct(
                    headers_to_remove = ["bar"]
                )
            ),
            deny_response = struct(
                status = struct(code = 7),
                denied_response = struct(
                    body = "no go"
                )
            )
        ))

    return struct(envoy_check_response = struct(
        status = struct(code = 0),
        ok_response = struct(
            headers_to_remove = ["foo"]
        )
    ))

def envoy_check(req):
    # Add code here to inspect the request and extract the information you need to pass to Cerbos

    # Return the Cerbos request and the corresponding Envoy response to return based on the decision from the PDP
    return struct(cerbos_mapping = struct(
        check_request = struct( (1)
            principal = struct(
                id = "daffy",
                roles = ["duck"]
            ),
            resources = [
                struct(
                    actions = ["GET"],
                    resource = struct(
                        id = "http",
                        kind = "request",
                        attr = {
                            "path": "/foo"
                        }
                    )
                )
            ]
        ),
        allow_response = struct( (2)
            status = struct(code = 0)
        ),
        deny_response = struct( (3)
            status = struct(code = 7),
            denied_response = struct(body = "Go away")
        )
    ))

def envoy_check(req): (1)
    # Add code here to inspect the request and extract the information you need to pass to Cerbos

    # Return the Cerbos request
    return struct(cerbos_check_request = struct(
        principal = struct(
            id = "daffy",
            roles = ["duck"]
        ),
        resources = [
            struct(
                actions = ["GET"],
                resource = struct(
                    id = "http",
                    kind = "request",
                    attr = {
                        "path": "/foo"
                    }
                )
            )
        ]
    ))

def map_cerbos_response(resp): (2)
    # Inspect the Cerbos response here and construct the appropriate Envoy response

    # Return the constructed Envoy response
    return struct(
        status = struct(code = 0),
        ok_response = struct(
            headers_to_remove = ["baz"]
        )
    )

struct(
    envoy_request = struct( ... ), (1)
    cerbos_request = struct( ... ), (2)
    cerbos_response = struct( ... ) (3)
)

def handle_augment(req):
    if hasattr(req, "principal"):
        if req.principal.id == "john":
            req.principal.attr = {
                "geography": "GB",
                "department": "IT"
            }

    return req

def augment_check_request(req):
    return handle_augment(req)

def augment_plan_request(req):
    return handle_augment(req)

apiVersion: api.cerbos.dev/v1
resourcePolicy:
  resource: "invoice"
  version: "default"
  rules:
    - actions: ["view"]
      effect: EFFECT_ALLOW
      roles: ["user"]
      condition:
        match:
          expr: P.attr.employee.department == R.attr.department

QUERY = "SELECT department, country, manager_email FROM employees WHERE id = :uid"

def enrich(req):
    if not hasattr(req, "principal") or req.principal.id == "":
        return req

    resp = cerbos.data_source_lookup( (1)
        datasource = "employees",
        query = QUERY,
        query_parameters = {"uid": req.principal.id}, (2)
        cache_key = "employee:" + req.principal.id, (3)
        cache_expiry = 5 * time.minute,
    )

    if resp == None or resp.result == None: (4)
        return req

    req.principal.attr["employee"] = resp.result
    return req

def augment_check_request(req):
    return enrich(req)

def augment_plan_request(req):
    return enrich(req)

def handle_http_route(req):
    # Add code here to process the request and determine a response.

    # Return the response
    return struct(http_response = struct(
        status = 401, (1)
        headers = { (2)
            "Content-Type": struct(values = ["application/text"])
        },
        body = "Access denied" (3)
    ))

def handle_http_route(req):
    # Add code here to examine the request and construct a Cerbos CheckResources request from it.

    # Return the Cerbos request and the HTTP response to send based on the Cerbos response.
    return struct(cerbos_mapping = struct(
        check_request = struct( (1)
            principal = struct(
                id = "daffy",
                roles = ["duck"]
            ),
            resources = [
                struct(
                    actions = ["GET"],
                    resource = struct(
                        id = "http",
                        kind = "request",
                        attr = {
                            "path": "/foo"
                        }
                    )
                )
            ]
        ),
        allow_response = struct( (2)
            status = 200,
            body = "Welcome"
        ),
        deny_response = struct( (3)
            status = 401,
            body = "Access denied"
        )
    ))

def handle_http_route(req):
    # Add code here to inspect the request and construct a Cerbos CheckResources request from it.

    # Return the Cerbos request constructed above
    return struct(check_request = struct(
        principal = struct(
            id = "daffy",
            roles = ["duck"]
        ),
        resources = [
            struct(
                actions = ["GET"],
                resource = struct(
                    id = "http",
                    kind = "request",
                    attr = {
                        "path": "/foo"
                    }
                )
            )
        ]
    ))

def handle_cerbos_response(resp):
    # Add code here to inspect the Cerbos response and construct the appropriate HTTP response

    # Return the constructed HTTP response
    return struct(http_response = struct(
        status = 401,
        headers = {
            "Content-Type": struct(values = ["application/text"])
        },
        body = "Access denied"
    ))

struct(
    http_request = struct( ... ), (1)
    cerbos_request = struct( ... ), (2)
    cerbos_response = struct( ... ) (3)
)

struct(
    status = 401,
    headers = {
        "Content-Type": struct(values = ["application/json"])
    },
    body = "Access denied"
)

load("oauth", "oauth")

def do_oauth_request():
    client = oauth.client_credentials_client(token_url=TOKEN_URL, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, scopes=[SCOPE], endpoint_params={"foo": "bar"}, persist_key=PERSIST_KEY)
    resp = client.get(AUTHENTICATED_ENDPOINT)
    return resp.status_code