Enforcement code sample

This Python function is the check from Check a token at your resource server with introspection, in code. It uses only the standard library. It reads POLICYARC_URL, RS_CLIENT_ID and RS_CLIENT_SECRET from the environment.

import base64
import json
import os
import sys
import urllib.parse
import urllib.request

POLICYARC_URL = os.environ["POLICYARC_URL"]
RS_CLIENT_ID = os.environ["RS_CLIENT_ID"]
RS_CLIENT_SECRET = os.environ["RS_CLIENT_SECRET"]


def introspect(token, resource_type, scope):
    context = {"resource_type": resource_type, "scopes": [scope]}
    body = urllib.parse.urlencode({"token": token, "context": json.dumps(context)}).encode()
    basic = base64.b64encode(f"{RS_CLIENT_ID}:{RS_CLIENT_SECRET}".encode()).decode()
    request = urllib.request.Request(
        f"{POLICYARC_URL}/introspect",
        data=body,
        headers={"Authorization": f"Basic {basic}", "Content-Type": "application/x-www-form-urlencoded"},
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return json.load(response)


def check(token, resource_type, scope):
    result = introspect(token, resource_type, scope)
    if not result.get("active"):
        status = 401 if result.get("inactive_reason") == "invalid_token" else 403
        return status, result.get("inactive_reason"), {}
    extensions = result.get("extensions", {})
    granted_types = {g.get("type") for g in extensions.get("granted_resources", [])}
    if scope not in (result.get("scope") or "").split() or resource_type not in granted_types:
        return 403, "not_granted_for_resource", {}
    return 200, None, extensions.get("obligations", {})


if __name__ == "__main__":
    print(check(sys.argv[1], sys.argv[2], sys.argv[3]))

Save it as enforce.py and run it with a token, a resource type and the scope the call needs.

python3 enforce.py "$TOKEN" urn:example:orders orders:read

What we saw with the orders:read token and the redact control from Obligations deployed.

Arguments	Output
"$TOKEN" urn:example:orders orders:read	(200, None, {'filter_words': {'replacement': '[REDACTED]', 'words': ['confidential']}})
"$TOKEN" urn:example:orders orders:write	(403, 'insufficient_scope', {})
nope urn:example:orders orders:read	(401, 'invalid_token', {})
A token for urn:example:x with scope orders:read, then urn:example:orders orders:read	(403, 'not_granted_for_resource', {})

The last row is why check() reads scope and granted_resources as well as active. Introspection answered active: true for that token, with an empty scope, as the known issue in Check the scope and resource describes. not_granted_for_resource is the sample's own reason, not one the server returns.

Your API serves the call only on 200, then applies each obligation it supports before it returns the response.