docs/development/integration-authoring.md
Rama main · réplica de lectura del repositorio original.
# Writing an integration
An *integration* is one product (github, jira). A *connection* is one configured system of that
product. This page is the contract every integration package follows. Paths are under
`toolpass-py/`. `src/toolpass/integrations/fake.py` and `src/toolpass/integrations/pagerduty/` are
the reference implementations.
## Package layout
```
src/toolpass/integrations/<name>/__init__.py exports INTEGRATION
src/toolpass/integrations/<name>/<name>.py the Integration and Connection classes
src/toolpass/integrations/<name>/actions.py action table, resource parsing
tests/integrations/<name>/test_<name>.py tests against a fake upstream
tests/integrations/<name>/test_fuzz.py property tests of the resource parser
docs/integrations/<name>.md
```
The package name equals the integration name (`googleworkspace`, `microsoft365`). The package
exposes one instance as `INTEGRATION`, and registration is one line: its name in `NAMES` in
`src/toolpass/integrations/__init__.py`. Integrations use only the standard library and toolpass's
own modules; signing with a private key goes through `toolpass.authx`, which imports
`cryptography` (the `crypto` extra) only when it is needed.
## The classes (`toolpass.core.integration`)
```python
class Integration(ABC):
def name(self) -> str: ... # lowercase, the "integration:" config value
def fields(self) -> list[Field]: ... # config keys beyond ca_file/tls_server_name/proxy_url/timeout
def actions(self) -> list[Action]: ... # fixed actions; Action(..., pattern=True) documents a shape
def new(self, ctx: Context, s: Settings, d: Deps) -> Connection: ...
def match_action(self, name: str) -> Action | None: ... # optional, for pattern actions
class Connection(ABC):
def resolve_identity(self, ctx: Context, u: User) -> Identity: ...
def check(self, ctx: Context, r: CheckRequest) -> Decision: ...
def probe(self, ctx: Context) -> ProbeResult: ...
```
- `new` must not touch the network. Build the transport with `d.http_client(s)` and wrap it in
`httpx.Client(http=..., base=..., auth=..., logger=d.logger)`. Read secrets at call time, in the
`auth` function, through `s.secret("credential").get_string()`, never in `new` (files rotate).
Raise `ValueError` from `new` for a setting that cannot work; `toolpass validate` reports it.
- `fields`: use `url_field`, `credential_field` and `connection_ref_field`; set `default`, `enum`
and `validate` on a `Field` where sensible. Secret fields are `secret=True` and arrive as `env:`
or `file:` references (`Settings.secret(name)`). A field with `code=True` takes a Python value
(a callable) given in `Toolpass(connections=[...])`, never from a file, read with
`Settings.object(name)` (None when absent); it cannot be secret, required or defaulted. The rest
are strings (`Settings.get`, `Settings.bool`). Never declare `url`, `credential` and the like through both a helper and a
literal. A `*_connection` field names another connection, which `d.connection(id)` returns. Its
`ref` is the integration it must be, or several separated by `|` (`"aws|okta"`), any of which is
accepted.
- An integration whose system grants by the groups of the user's login declares
`*oidc_fields("<the system>")` and builds `core.oidc.Verifier.from_settings(s, d)` in `new`; in
`resolve_identity`, when the caller sent no groups and the verifier is set, `token_groups(ctx, u)`
reads them from the ID token the check carries (`User.token`). argocd shows the pattern. Nothing
provider-specific belongs there: the issuer, audience and claim names are the connection's settings.
- A directory or cloud account that knows a user's groups can implement
`KubernetesSubjectSource` (`kubernetes_subject(ctx, email, cluster)` and the
`kubernetes_needs_cluster` attribute), so a kubernetes connection can name it as its
`groups_connection`; add the integration to the kubernetes `GROUP_SOURCES`.
- Pattern actions (`raw:<verb>:<resource>`) are listed with `pattern=True` and matched by
`match_action`.
A minimal integration, for a made-up API:
```python
import re
from toolpass.core import jsonx
from toolpass.core.catalog import Action
from toolpass.core.decision import Code, allowed, denied, errorf, unknown_decision, user_ambiguous, user_not_found, wrap_error
from toolpass.core.integration import Connection, Identity, Integration, ProbeResult, credential_field, url_field
from toolpass.core.secret import SecretError
from toolpass.net import httpx
_PROJECT_KEY = re.compile(r"[A-Z][A-Z0-9]{1,9}")
class Acme(Integration):
def name(self):
return "acme"
def fields(self):
return [url_field(True, "the account URL"), credential_field(True, "a read-only API token")]
def actions(self):
return [Action("project.write", "change a project's settings")]
def new(self, ctx, s, d):
cred = s.secret("credential")
def auth(ctx, r):
try:
token = cred.get_string()
except SecretError as e:
raise wrap_error(Code.CREDENTIAL_REJECTED, e, "the API token could not be read") from e
r.headers.set("Authorization", "Bearer " + token)
return AcmeConnection(httpx.Client(http=d.http_client(s), base=s.get("url").rstrip("/"), auth=auth, logger=d.logger))
class AcmeConnection(Connection):
def __init__(self, api):
self.api = api
def resolve_identity(self, ctx, u):
try:
_, body = self.api.get_json(ctx, "/api/users", {"email": u.email})
users = [jsonx.obj(x) for x in jsonx.arr(jsonx.obj(body), "users")]
except Exception as e:
raise httpx.classify(e) from e
if not users:
raise user_not_found(f"no Acme user has email {u.email}")
if len(users) > 1:
raise user_ambiguous(f"{len(users)} Acme users have email {u.email}")
return Identity(id=jsonx.s(users[0], "id"), display=u.email)
def check(self, ctx, r):
if r.resource.type != "project" or not _PROJECT_KEY.fullmatch(r.resource.id):
raise errorf(Code.INVALID_REQUEST, "resource must be project:<KEY>")
path = f"/api/projects/{httpx.path_escape(r.resource.id)}/members/{httpx.path_escape(r.identity.id)}"
try:
_, body = self.api.get_json(ctx, path)
except Exception as e:
if httpx.status(e) == 404:
return unknown_decision(Code.RESOURCE_NOT_VISIBLE, f"project {r.resource.id} does not exist or toolpass cannot see it")
raise httpx.classify(e) from e
role = jsonx.s(jsonx.obj(body), "role")
if role in ("admin", "maintainer"):
return allowed(f"{r.identity.display} is a {role} of project {r.resource.id}")
return denied(f"{r.identity.display} is a {role or 'non-member'} of project {r.resource.id}, not a maintainer")
def probe(self, ctx):
try:
_, body = self.api.get_json(ctx, "/api/me")
except Exception as e:
raise httpx.classify(e) from e
return ProbeResult(summary="authenticated as " + jsonx.s(jsonx.obj(body), "name"))
INTEGRATION = Acme() # in the package's __init__.py
```
The real integrations are fully typed (`mypy --strict` runs on `src/toolpass`); the types are left
out here for space.
## Decisions
`deny` only when the third-party system positively says no. Everything toolpass could not evaluate
is `unknown` with a code from `toolpass.core.decision.Code`:
| Situation | Return or raise |
|---|---|
| allowed | `return allowed("...")` |
| positively refused | `return denied("...")` |
| no account for the email | `raise user_not_found(...)` from `resolve_identity` |
| several accounts | `raise user_ambiguous(...)` |
| toolpass's credential rejected (401, or 403 where it means "toolpass lacks the right") | `raise wrap_error(Code.CREDENTIAL_REJECTED, err, "...")` |
| resource not visible / not found (404 that may mean "no access") | `return unknown_decision(Code.RESOURCE_NOT_VISIBLE, ...)` |
| a policy construct toolpass does not model | `return unsupported(...)` |
| bad resource shape for this integration | `raise errorf(Code.INVALID_REQUEST, ...)` |
| transport, 5xx, 429, timeout | `raise httpx.classify(err) from err` |
Any other exception that escapes `resolve_identity` or `check` becomes `unknown` with
`upstream_error` (or `upstream_timeout`), so a bug fails closed; but classify what you expect, so
that a 401 says `credential_rejected`. Decision text must never contain a secret or an upstream
body. Keep it one sentence.
## Upstream JSON
Read decoded bodies with `toolpass.core.jsonx`: `jsonx.obj(v)`, and `jsonx.s`, `i`, `b`, `arr`,
`o` and `strs` for a key. A missing key or `null` is the zero value; a value of the wrong type
raises `jsonx.DecodeError`, so a malformed answer is `upstream_error`, never a silently wrong
decision. Do not index decoded JSON directly.
## Resources
`type:id`, parsed by `toolpass.core.catalog.parse_resource` into `r.resource.type` and
`r.resource.id` (and query parameters, for forms such as `namespace:payments?name=api`). The
integration validates the type and id with strict regular expressions (`fullmatch`) before putting
either into a URL path (`httpx.path_escape`), a query, a GraphQL variable, a JQL or SOQL string, or
anything else. Git hosts use `catalog.split_branch` for `@branch`.
## HTTP
`httpx.Client` handles TLS, retries (idempotent calls only), 429 with `Retry-After`, the body cap
and redaction. Use `get_json`, `post_json`, `do`, `paginate` and `next_link` (which refuses a next
page on another host, so the credential never travels there). Bodies are never logged. Use
`httpx.status(err)` to branch on 403 and 404, and `httpx.classify(err)` for everything else. Every
call takes the `ctx` toolpass passed in: it carries the connection's timeout and cancellation.
Every response the client completes during `resolve_identity` and `check` is recorded as evidence on
the decision (method, path, status, and the `ETag` or the body's SHA-256) and written to the
decision log by the engine; an integration does nothing for this. Calls made from the client's
`auth` function or by an `authx` token source are not recorded. A lookup an integration caches in a
`toolpass.core.cache.TTL` (role definitions, policies) is replayed, marked `cached`, on every check
the entry serves, and is looked up again for a check with `"fresh": true`
(`toolpass.core.evidence.fresh(ctx)`). Keep such lookups in a `TTL` rather than a hand-rolled dict
so both hold. A map read from a local file (`role_map_file`, `user_map_file`) is configuration, not
upstream state, and keeps its own re-read schedule.
## Auth (`toolpass.authx`)
`authx.token.TokenSource` caches a token and refreshes it 5 minutes before expiry.
`authx.jwt.sign_jwt` signs RS256 and PS256 with `kid` and `x5t#S256` headers.
`authx.oauth2.client_credentials`, `client_assertion` and `jwt_bearer` are token-endpoint fetchers,
and `authx.oauth2.classify_token_error` maps their failures. `authx.sigv4` signs AWS requests and
`authx.google` reads Google service-account keys and the metadata server.
## Tests
Use the harness in `tests/harness` (`from tests import harness as itest`):
- The `srv` fixture is an `itest.Server`: a TLS fake upstream with `handle`, `json`, `calls`,
`last_call` and `fail(mode)` injection. Handlers take `(w, r)`.
- `itest.deps(srv)` returns `Deps` trusting the server, and the logs. After every test, the
conftest fails it if `itest.CANARY` appears in any log line. Every test secret is
`itest.literal("...")`, which carries the canary. Put the canary in fake upstream bodies too.
- `itest.settings(id, integration, values, secrets)` builds `Settings` with a short timeout.
- `itest.check(conn, integ, user, action, resource)` runs the check like the engine;
`itest.expect_code(d, code)` asserts the code.
- `itest.failure_cases(srv, lambda: ...)` asserts the `unknown` codes for 500, 429, 401 and a
timeout.
- `srv.use_spec(spec_from_env("<name>"), SpecOptions(...))` (from `tests.harness.spec`) validates
every request against the vendor's published API description (OpenAPI, Google discovery,
botocore or GraphQL SDL) when `TOOLPASS_SPECS_DIR` holds `<name>.spec`; `test/specs/fetch.sh`
downloads them. A wrong path, method, missing required parameter or body field fails the test.
Use `strip_prefix` for gateway prefixes, `ignore_paths` for endpoints the description lacks (say
why in a comment), and `allow_query` and `optional_params` for parameters newer or older than the
description.
```python
from toolpass.core.context import background
from toolpass.core.decision import Code
from toolpass.core.integration import User
from toolpass.integrations.acme import Acme
from tests import harness as itest
from tests.harness.spec import SpecOptions, spec_from_env
def setup(srv):
srv.use_spec(spec_from_env("acme"), SpecOptions())
srv.json("GET", "/api/users", 200, {"users": [{"id": "u1", "email": "dana@example.com"}]})
srv.json("GET", "/api/projects/PAY/members/u1", 200, {"role": "maintainer"})
deps, _ = itest.deps(srv)
s = itest.settings("acme-main", "acme", {"url": srv.url}, {"credential": itest.literal("token")})
return Acme().new(background(), s, deps)
def test_action_project_write_allow(srv):
d = itest.check(setup(srv), Acme(), User(email="dana@example.com"), "project.write", "project:PAY")
itest.expect_code(d, Code.ALLOWED)
def test_action_project_write_deny(srv):
c = setup(srv)
srv.json("GET", "/api/projects/PAY/members/u1", 200, {"role": "viewer"})
itest.expect_code(itest.check(c, Acme(), User(email="dana@example.com"), "project.write", "project:PAY"), Code.DENIED)
def test_failures(srv):
c = setup(srv)
itest.failure_cases(srv, lambda: itest.check(c, Acme(), User(email="dana@example.com"), "project.write", "project:PAY"))
```
The coverage gate, `tests/integrations/test_registry.py`, requires for every non-pattern action of
every registered integration a `test_action_<name>_allow` and a `test_action_<name>_deny` among the
integration's tests, with every character outside `[A-Za-z0-9]` in the name replaced by `_`
(`repo.push` -> `test_action_repo_push_allow`; case does not matter). It also requires the module
name to equal the integration name.
Every resource parser has property tests in `test_fuzz.py` (Hypothesis) asserting that an accepted
resource is made only of validated pieces before it reaches a URL or query.
## Docs
`docs/integrations/<name>.md` has, in this order: how it checks (one paragraph), the credential to
create and the minimum permissions, the connection keys, resources, actions, decisions table, what it
cannot see, an **Unverified** section listing every behaviour marked `# UNVERIFIED:` in the code, and
how to test. Add the integration to the table in `docs/integrations/README.md`.
## Words
Integration, connection. Not instance, adapter, target, caller, cluster/server/site as a config key.
The endpoint is `/check`.