id: openfga-authorization-model-gate
namespace: company.team
description: |
Compare a proposed OpenFGA authorization model with a baseline on the same fixture tuples.
Assert expected allow and deny decisions, report changed decisions, and block a bad candidate.
All test writes use a new temporary store; production promotion is an optional approved step.
concurrency: { limit: 1, behavior: QUEUE }
triggers:
- id: model_change_webhook
type: io.kestra.plugin.core.trigger.Webhook
key: openfga-model-gate-change-me
description: "CI can POST to
/api/v1/main/executions/webhook/company.team/openfga-authorization-model-\
gate/openfga-model-gate-change-me after changing the key."
- id: nightly_model_check
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 2 * * *"
disabled: true
description: Enable this backstop only after replacing the example fixture and
webhook key with your maintained policy tests.
inputs:
- id: openfga_api_url
type: STRING
defaults: http://openfga-gate:8080
description: OpenFGA API base URL; the default reaches the local Docker service
used in the quick start.
- id: baseline_model_json
type: STRING
defaults: |-
{"schema_version":"1.1","type_definitions":[{"type":"user"},{"type":"document","relations":{"editor":{"this":{}},"viewer":{"union":{"child":[{"this":{}},{"computedUserset":{"relation":"editor"}}]}}},"metadata":{"relations":{"editor":{"directly_related_user_types":[{"type":"user"}]},"viewer":{"directly_related_user_types":[{"type":"user"}]}}}}]}
description: Baseline model in OpenFGA WriteAuthorizationModel JSON form;
replace it with the currently deployed model export.
- id: candidate_model_json
type: STRING
defaults: |-
{"schema_version":"1.1","type_definitions":[{"type":"user"},{"type":"document","relations":{"editor":{"this":{}},"viewer":{"this":{}}},"metadata":{"relations":{"editor":{"directly_related_user_types":[{"type":"user"}]},"viewer":{"directly_related_user_types":[{"type":"user"}]}}}}]}
description: Proposed model in OpenFGA WriteAuthorizationModel JSON form; this
is written only in the temporary store.
- id: fixture_tuples_json
type: STRING
defaults: |-
[{"user":"user:alice","relation":"viewer","object":"document:plan"},{"user":"user:bob","relation":"editor","object":"document:plan"}]
description: Relationship tuples used under both models; use a representative
and sanitized fixture, never a production tuple dump.
- id: assertions_json
type: STRING
defaults: |-
[{"name":"direct_viewer","user":"user:alice","relation":"viewer","object":"document:plan","expected_baseline":true,"expected_candidate":true},{"name":"editor_keeps_view","user":"user:bob","relation":"viewer","object":"document:plan","expected_baseline":true,"expected_candidate":true},{"name":"stranger_denied","user":"user:mallory","relation":"viewer","object":"document:plan","expected_baseline":false,"expected_candidate":false}]
description: Named allow/deny checks with expected decisions for both models;
add every security-sensitive case before using the gate in CI.
- id: promote_to_production
type: BOOL
defaults: false
description: Require human approval before writing the passing candidate to a
production store; the default performs no production write.
- id: production_store_id
type: STRING
defaults: disabled
description: Target production store ID for an approved promotion; replace the
sentinel only when promotion is enabled.
- id: expected_production_model_id
type: STRING
defaults: disabled
description: Current production model ID pinned by the change request; a
mismatch blocks approval or a late write.
tasks:
- id: validate_fixture
type: io.kestra.plugin.scripts.python.Script
description: Reject malformed models, tuples, checks, and API URLs before
creating any OpenFGA store.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
openfga_api_url.txt: "{{ inputs.openfga_api_url }}"
baseline_model.json: "{{ inputs.baseline_model_json }}"
candidate_model.json: "{{ inputs.candidate_model_json }}"
fixture_tuples.json: "{{ inputs.fixture_tuples_json }}"
assertions.json: "{{ inputs.assertions_json }}"
promotion.txt: "{{ inputs.promote_to_production }}"
production_store_id.txt: "{{ inputs.production_store_id }}"
expected_production_model_id.txt: "{{ inputs.expected_production_model_id }}"
script: |
import json
from urllib.parse import urlsplit
def load(name):
with open(name, encoding="utf-8") as file:
return json.load(file)
with open("openfga_api_url.txt", encoding="utf-8") as file:
api_url = file.read().strip()
url = urlsplit(api_url)
if (url.scheme not in ("http", "https") or not url.hostname or
url.username or url.password or url.path not in ("", "/") or
url.query or url.fragment):
raise ValueError("openfga_api_url must be an HTTP(S) origin without credentials, path, or query")
for name in ("baseline_model.json", "candidate_model.json"):
model = load(name)
if (not isinstance(model, dict) or model.get("schema_version") != "1.1" or
not isinstance(model.get("type_definitions"), list) or
not model["type_definitions"]):
raise ValueError(name + " must be an OpenFGA 1.1 model with type_definitions")
tuples = load("fixture_tuples.json")
checks = load("assertions.json")
if not isinstance(tuples, list) or not 1 <= len(tuples) <= 100:
raise ValueError("fixture_tuples_json must contain 1 to 100 tuples")
if not isinstance(checks, list) or not 1 <= len(checks) <= 100:
raise ValueError("assertions_json must contain 1 to 100 checks")
tuple_fields = ("user", "relation", "object")
for number, entry in enumerate(tuples):
if not isinstance(entry, dict) or any(not isinstance(entry.get(key), str) or not entry[key] for key in tuple_fields):
raise ValueError(f"fixture tuple {number} needs nonempty user, relation, and object")
if len({tuple(entry[key] for key in tuple_fields) for entry in tuples}) != len(tuples):
raise ValueError("fixture tuples must be unique")
names = set()
for number, entry in enumerate(checks):
if not isinstance(entry, dict) or any(not isinstance(entry.get(key), str) or not entry[key] for key in ("name", *tuple_fields)):
raise ValueError(f"assertion {number} needs a name, user, relation, and object")
if type(entry.get("expected_baseline")) is not bool or type(entry.get("expected_candidate")) is not bool:
raise ValueError(f"assertion {number} needs boolean expected_baseline and expected_candidate")
if entry["name"] in names:
raise ValueError("assertion names must be unique")
names.add(entry["name"])
with open("promotion.txt", encoding="utf-8") as file:
promote = file.read().strip().lower() == "true"
if promote:
for name in ("production_store_id.txt", "expected_production_model_id.txt"):
with open(name, encoding="utf-8") as file:
identifier = file.read().strip()
if len(identifier) != 26 or not identifier.isalnum() or identifier.upper() != identifier:
raise ValueError(name + " must contain an OpenFGA ULID when promotion is enabled")
print("::" + json.dumps({"outputs": {"tuple_count": len(tuples), "check_count": len(checks)}}) + "::")
- id: compare_models
type: io.kestra.plugin.scripts.python.Script
description: Create a scratch store, write both immutable models and shared
fixtures, compare pinned checks, then delete the store.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
openfga_api_url.txt: "{{ inputs.openfga_api_url }}"
baseline_model.json: "{{ inputs.baseline_model_json }}"
candidate_model.json: "{{ inputs.candidate_model_json }}"
fixture_tuples.json: "{{ inputs.fixture_tuples_json }}"
assertions.json: "{{ inputs.assertions_json }}"
env:
OPENFGA_API_TOKEN: "{{ secret('OPENFGA_API_TOKEN') }}"
script: |
import json
import hashlib
import os
import time
import urllib.error
import urllib.request
from uuid import uuid4
def load(name):
with open(name, encoding="utf-8") as file:
return json.load(file)
with open("openfga_api_url.txt", encoding="utf-8") as file:
origin = file.read().strip().rstrip("/")
token = os.environ.get("OPENFGA_API_TOKEN", "")
def call(method, path, body=None, retry_safe=False):
payload = None if body is None else json.dumps(body, separators=(",", ":")).encode()
headers = {"Accept": "application/json"}
if payload is not None:
headers["Content-Type"] = "application/json"
if token:
headers["Authorization"] = "Bearer " + token
attempts = 3 if retry_safe else 1
for attempt in range(attempts):
request = urllib.request.Request(origin + path, data=payload, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=10) as response:
raw = response.read()
parsed = json.loads(raw) if raw else {}
if not isinstance(parsed, dict) or parsed.get("error") or parsed.get("code"):
raise RuntimeError(f"OpenFGA {method} {path} returned an API error: {parsed}")
return parsed
except urllib.error.HTTPError as error:
detail = error.read(512).decode("utf-8", "replace")
if method == "DELETE" and error.code == 404:
return {}
if retry_safe and error.code in (429, 502, 503, 504) and attempt + 1 < attempts:
time.sleep(2 ** attempt)
continue
raise RuntimeError(f"OpenFGA {method} {path} failed with HTTP {error.code}: {detail}") from error
except (urllib.error.URLError, TimeoutError) as error:
if retry_safe and attempt + 1 < attempts:
time.sleep(2 ** attempt)
continue
raise RuntimeError(f"OpenFGA {method} {path} could not connect: {error.reason if hasattr(error, 'reason') else error}") from error
store_id = None
try:
store = call("POST", "/stores", {"name": "kestra-model-gate-" + uuid4().hex})
store_id = store.get("id")
if not isinstance(store_id, str) or not store_id:
raise RuntimeError("OpenFGA create-store response lacked id; inspect the service for an orphaned temporary store")
prefix = "/stores/" + store_id
baseline = call("POST", prefix + "/authorization-models", load("baseline_model.json"))
baseline_id = baseline.get("authorization_model_id")
if not isinstance(baseline_id, str) or not baseline_id:
raise RuntimeError("OpenFGA baseline model write returned no authorization_model_id")
baseline_read = call("GET", prefix + "/authorization-models/" + baseline_id, retry_safe=True)
normalized_baseline = baseline_read.get("authorization_model")
if not isinstance(normalized_baseline, dict) or normalized_baseline.get("id") != baseline_id:
raise RuntimeError("OpenFGA could not read back the baseline model")
baseline_fingerprint = hashlib.sha256(json.dumps({
key: value for key, value in normalized_baseline.items() if key != "id"
}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
call("POST", prefix + "/write", {
"authorization_model_id": baseline_id,
"writes": {"tuple_keys": load("fixture_tuples.json")}
})
candidate = call("POST", prefix + "/authorization-models", load("candidate_model.json"))
candidate_id = candidate.get("authorization_model_id")
if not isinstance(candidate_id, str) or not candidate_id or candidate_id == baseline_id:
raise RuntimeError("OpenFGA candidate model write returned an invalid authorization_model_id")
results = []
changes = []
for check in load("assertions.json"):
tuple_key = {key: check[key] for key in ("user", "relation", "object")}
decisions = {}
for label, model_id in (("baseline", baseline_id), ("candidate", candidate_id)):
response = call("POST", prefix + "/check", {
"authorization_model_id": model_id,
"tuple_key": tuple_key,
"consistency": "HIGHER_CONSISTENCY"
}, retry_safe=True)
if type(response.get("allowed")) is not bool:
raise RuntimeError("OpenFGA check response lacked a boolean allowed value")
decisions[label] = response["allowed"]
result = {"name": check["name"], "tuple_key": tuple_key,
"baseline": decisions["baseline"], "candidate": decisions["candidate"],
"expected_baseline": check["expected_baseline"],
"expected_candidate": check["expected_candidate"],
"passed": decisions["baseline"] == check["expected_baseline"] and
decisions["candidate"] == check["expected_candidate"]}
results.append(result)
if decisions["baseline"] != decisions["candidate"]:
changes.append({"name": check["name"], "tuple_key": tuple_key,
"from": decisions["baseline"], "to": decisions["candidate"]})
report = {"passed": all(result["passed"] for result in results),
"checked": len(results), "failed": sum(not result["passed"] for result in results),
"changed": changes, "assertions": results,
"baseline_model_id": baseline_id, "candidate_model_id": candidate_id,
"baseline_model_sha256": baseline_fingerprint,
"temporary_store_id": store_id}
print("::" + json.dumps({"outputs": {"report": report, "passed": report["passed"]}}) + "::")
print(f"Checked {len(results)} decisions under each model; {len(changes)} changed, {report['failed']} assertion failures.")
finally:
if store_id:
try:
call("DELETE", "/stores/" + store_id, retry_safe=True)
print("Deleted temporary OpenFGA store " + store_id)
except Exception as error:
raise RuntimeError(f"Temporary store {store_id} cleanup failed; delete it manually: {error}") from error
- id: enforce_gate
type: io.kestra.plugin.core.flow.If
description: A mismatch is an intentional deployment block; operational failures
use the global error handler.
condition: "{{ outputs.compare_models.vars.passed == false }}"
then:
- id: log_blocked_decisions
type: io.kestra.plugin.core.log.Log
message: "Candidate blocked: {{ outputs.compare_models.vars.report.failed }} of
{{ outputs.compare_models.vars.report.checked }} assertions failed.
Inspect compare_models report before promotion."
- id: block_candidate
type: io.kestra.plugin.core.execution.Exit
state: FAILED
else:
- id: log_approved_decisions
type: io.kestra.plugin.core.log.Log
message: "Candidate passed {{ outputs.compare_models.vars.report.checked }}
assertions; {{ outputs.compare_models.vars.report.changed | length }}
tested decisions changed as expected. Review the report before any
production promotion."
- id: consider_production_promotion
type: io.kestra.plugin.core.flow.If
description: Production remains untouched unless the caller enables promotion
and a human approves it.
condition: "{{ inputs.promote_to_production == true }}"
then:
- id: check_production_revision
type: io.kestra.plugin.scripts.python.Script
description: Confirm that the live latest model ID and normalized model content
match the tested baseline before asking for approval.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
openfga_api_url.txt: "{{ inputs.openfga_api_url }}"
production_store_id.txt: "{{ inputs.production_store_id }}"
expected_production_model_id.txt: "{{ inputs.expected_production_model_id }}"
baseline_sha256.txt: "{{ outputs.compare_models.vars.report.baseline_model_sha256 }}"
env:
OPENFGA_API_TOKEN: "{{ secret('OPENFGA_API_TOKEN') }}"
script: |
import hashlib
import json
import os
import time
import urllib.error
import urllib.request
def read_text(name):
with open(name, encoding="utf-8") as file:
return file.read().strip()
origin = read_text("openfga_api_url.txt").rstrip("/")
store_id = read_text("production_store_id.txt")
expected = read_text("expected_production_model_id.txt")
fingerprint = read_text("baseline_sha256.txt")
headers = {"Accept": "application/json"}
token = os.environ.get("OPENFGA_API_TOKEN", "")
if token:
headers["Authorization"] = "Bearer " + token
url = origin + "/stores/" + store_id + "/authorization-models?page_size=1"
for attempt in range(3):
try:
with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=10) as response:
result = json.load(response)
break
except urllib.error.HTTPError as error:
if error.code not in (429, 502, 503, 504) or attempt == 2:
raise RuntimeError(f"Production model read failed with HTTP {error.code}") from error
except (urllib.error.URLError, TimeoutError) as error:
if attempt == 2:
raise RuntimeError(f"Production model read could not connect: {error}") from error
time.sleep(2 ** attempt)
models = result.get("authorization_models") if isinstance(result, dict) else None
if not isinstance(models, list) or len(models) != 1 or not isinstance(models[0], dict):
raise RuntimeError("Production latest-model response was missing a model")
latest = models[0]
if latest.get("id") != expected:
raise RuntimeError(f"Production model changed before approval: expected {expected}, latest {latest.get('id')}")
actual = hashlib.sha256(json.dumps({
key: value for key, value in latest.items() if key != "id"
}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
if actual != fingerprint:
raise RuntimeError("The supplied baseline differs from the current production model; update fixtures and rerun")
print("::" + json.dumps({"outputs": {"latest_model_id": expected}}) + "::")
- id: describe_approval
type: io.kestra.plugin.core.log.Log
message: "Approval requested for production store {{ inputs.production_store_id
}}. Candidate passed {{ outputs.compare_models.vars.report.checked }}
checks; latest model {{
outputs.check_production_revision.vars.latest_model_id }} matched the
tested baseline. Review the decision diff and candidate model before
resuming."
- id: request_approval
type: io.kestra.plugin.core.flow.Pause
description: Approve the tested candidate for the pinned production store and
model revision.
onResume:
- id: approved
type: BOOL
defaults: false
description: Explicitly approve the production model write after reviewing the
report and candidate.
- id: approver_note
type: STRING
required: false
description: Record the change ticket or reason for the approval decision.
- id: require_approval
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.request_approval.onResume.approved == true }}"
then:
- id: write_production_model
type: io.kestra.plugin.scripts.python.Script
description: Re-check the latest production revision after approval, write the
candidate once, and verify its new model ID.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
openfga_api_url.txt: "{{ inputs.openfga_api_url }}"
production_store_id.txt: "{{ inputs.production_store_id }}"
expected_production_model_id.txt: "{{ outputs.check_production_revision.vars.latest_model_id }}"
candidate_model.json: "{{ inputs.candidate_model_json }}"
env:
OPENFGA_API_TOKEN: "{{ secret('OPENFGA_API_TOKEN') }}"
script: |
import json
import os
import time
import urllib.error
import urllib.request
def read_text(name):
with open(name, encoding="utf-8") as file:
return file.read().strip()
origin = read_text("openfga_api_url.txt").rstrip("/")
store_id = read_text("production_store_id.txt")
expected = read_text("expected_production_model_id.txt")
endpoint = origin + "/stores/" + store_id + "/authorization-models"
headers = {"Accept": "application/json", "Content-Type": "application/json"}
token = os.environ.get("OPENFGA_API_TOKEN", "")
if token:
headers["Authorization"] = "Bearer " + token
def latest():
for attempt in range(3):
try:
with urllib.request.urlopen(urllib.request.Request(endpoint + "?page_size=1", headers=headers), timeout=10) as response:
payload = json.load(response)
models = payload.get("authorization_models") if isinstance(payload, dict) else None
if not isinstance(models, list) or len(models) != 1:
raise RuntimeError("Production latest-model response was missing a model")
return models[0].get("id")
except urllib.error.HTTPError as error:
if error.code not in (429, 502, 503, 504) or attempt == 2:
raise RuntimeError(f"Production model read failed with HTTP {error.code}") from error
except (urllib.error.URLError, TimeoutError) as error:
if attempt == 2:
raise RuntimeError(f"Production model read could not connect: {error}") from error
time.sleep(2 ** attempt)
current = latest()
if current != expected:
raise RuntimeError(f"Production model changed while approval was pending: expected {expected}, latest {current}; rerun the gate")
with open("candidate_model.json", encoding="utf-8") as file:
model = json.load(file)
request = urllib.request.Request(endpoint, data=json.dumps(model).encode(), headers=headers, method="POST")
try:
with urllib.request.urlopen(request, timeout=10) as response:
result = json.load(response)
except urllib.error.HTTPError as error:
detail = error.read(512).decode("utf-8", "replace")
raise RuntimeError(f"Production model write failed with HTTP {error.code}: {detail}") from error
except (urllib.error.URLError, TimeoutError) as error:
raise RuntimeError("Production write result is uncertain; inspect the latest model before retrying") from error
new_id = result.get("authorization_model_id") if isinstance(result, dict) else None
if not isinstance(new_id, str) or not new_id:
raise RuntimeError("Production write response lacked a model ID; inspect the latest model before retrying")
if latest() != new_id:
raise RuntimeError("Production latest model changed after the write; inspect concurrent writers")
print("::" + json.dumps({"outputs": {"production_model_id": new_id}}) + "::")
else:
- id: log_rejected_approval
type: io.kestra.plugin.core.log.Log
message: "Production promotion declined for store {{ inputs.production_store_id
}}. The candidate was not written."
- id: block_unapproved_write
type: io.kestra.plugin.core.execution.Exit
state: FAILED
errors:
- id: alert_operational_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Send one actionable alert for validation, OpenFGA API, or cleanup failure.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
OpenFGA model gate execution {{ execution.id }} failed in {{
tasksWithState('FAILED')[0]['taskId'] }}. Cause: {{ (errorLogs() | toJson
| jq('[.[] | .message | select(contains("Error:"))] | first') | first) ??
errorLogs()[0]['message'] }}. If a production write timed out, inspect the
latest model ID before retrying. If temporary-store cleanup failed, delete
the named store. Do not promote the candidate.
outputs:
- id: gate_report
type: JSON
description: Per-check baseline/candidate decisions and expected changes on
successful runs; inspect compare_models task outputs when blocked.
value: "{{ (outputs.compare_models.vars.report ?? {}) | toJson }}"
- id: production_model_id
type: STRING
description: New production model ID after explicit approval, or empty when the
flow only tests a candidate.
value: "{{ outputs.write_production_model.vars.production_model_id ?? '' }}"
