tegwick
a769024c94
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 8m30s
Preserve the four unpublished T02 checkpoints with identical implementation and test contents. Represent Kubernetes resource UIDs as explicit kind/resource_uid metadata so they are not mistaken for credential values. Standard staged-blob scanner clean; full repository gate 764 passed. T02 remains open for received authentication and native return acceptance. Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a198ed-70cc-7362-a569-d5dbef39b86a
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare or explicitly send the two fixed T02 public registration requests."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import mail_return_operation as operation
|
|
from selected_mailbox_check import TESTS
|
|
|
|
ENDPOINT = "https://users.coulomb.social/api/v1/public/registrations"
|
|
|
|
|
|
def prepare():
|
|
return {
|
|
"endpoint": ENDPOINT,
|
|
"source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
|
|
"requests": [
|
|
{
|
|
"idempotency_key": "csoc-t02-" + secrets.token_hex(24),
|
|
"body": {
|
|
"username": username,
|
|
"email": email,
|
|
"display_name": display,
|
|
"client_id": "coulomb-social",
|
|
"tenant": "tenant:coulomb",
|
|
},
|
|
}
|
|
for (username, email), display in zip(
|
|
TESTS.items(), ("T02 delivery check", "T02 return check"), strict=True
|
|
)
|
|
],
|
|
}
|
|
|
|
|
|
def send(packet, packet_path, approve, opener=urllib.request.urlopen):
|
|
operation.require(operation.digest(packet) == approve, "exact_probe_approval_required")
|
|
expected = prepare()
|
|
operation.require(
|
|
packet["endpoint"] == ENDPOINT
|
|
and packet["source_sha256"] == expected["source_sha256"]
|
|
and len(packet["requests"]) == 2,
|
|
"probe_source_or_scope_changed",
|
|
)
|
|
for actual, fixed in zip(packet["requests"], expected["requests"], strict=True):
|
|
operation.require(
|
|
actual["body"] == fixed["body"] and len(actual["idempotency_key"]) == 57,
|
|
"probe_scope_changed",
|
|
)
|
|
# A lost response leaves an explicit intent; inspect the authoritative ledger.
|
|
operation.durable(
|
|
packet_path.with_suffix(".intent.json"), operation.canonical({"digest": approve})
|
|
)
|
|
results = []
|
|
for index, item in enumerate(packet["requests"]):
|
|
request = urllib.request.Request(
|
|
ENDPOINT,
|
|
data=operation.canonical(item["body"]),
|
|
method="POST",
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Idempotency-Key": item["idempotency_key"],
|
|
},
|
|
)
|
|
try:
|
|
with opener(request, timeout=45) as response:
|
|
body = response.read(4096)
|
|
operation.require(
|
|
response.status == 202
|
|
and json.loads(body) == {"status": "verification_requested"},
|
|
"probe_outcome_unconfirmed",
|
|
)
|
|
result = {
|
|
"request_index": index,
|
|
"status": "verification_requested",
|
|
"http_status": 202,
|
|
}
|
|
except Exception:
|
|
result = {"request_index": index, "status": "unconfirmed_inspect_before_retry"}
|
|
operation.durable(
|
|
packet_path.with_suffix(".request-" + str(index) + ".json"), operation.canonical(result)
|
|
)
|
|
results.append(result)
|
|
if result["status"] != "verification_requested":
|
|
break
|
|
return {"requests": results, "identity_provisioned": False, "retry_automatically": False}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("action", choices=("prepare", "send"))
|
|
parser.add_argument("--packet", required=True, type=Path)
|
|
parser.add_argument("--approve")
|
|
args = parser.parse_args()
|
|
operation.private_directory(args.packet.parent)
|
|
if args.action == "prepare":
|
|
packet = prepare()
|
|
operation.durable(args.packet, operation.canonical(packet))
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"status": "probe_prepared",
|
|
"digest": operation.digest(packet),
|
|
"requests": [r["body"] for r in packet["requests"]],
|
|
"mail_sent": False,
|
|
}
|
|
)
|
|
)
|
|
else:
|
|
print(json.dumps(send(operation.read_json(args.packet), args.packet, args.approve)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as error:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"status": "probe_refused_or_unconfirmed",
|
|
"error_type": type(error).__name__,
|
|
"retry_automatically": False,
|
|
}
|
|
)
|
|
)
|
|
sys.exit(1)
|