User Story #53 » 0017-feat-gateway-publish-the-mesh-CA-public-key-for-fetc.patch
| app/com/aether/tome/api/routes/mesh.py | ||
|---|---|---|
|
|
||
|
# Vendor imports
|
||
|
import base64
|
||
|
import hashlib
|
||
|
from http.client import BAD_REQUEST, CONFLICT, FORBIDDEN, INTERNAL_SERVER_ERROR, NOT_FOUND, SERVICE_UNAVAILABLE
|
||
|
import logging
|
||
|
import json
|
||
| ... | ... | |
|
return resp_ok({"requests": pending_role_requests(UUID(m_id))})
|
||
|
|
||
|
|
||
|
@mesh_bp.route("/ca", methods=["GET"])
|
||
|
def mesh_ca_public_key(): # A1 trust-anchor publish
|
||
|
"""Publish the mesh enrollment CA's Ed25519 public trust anchor so a device can FETCH and PIN
|
||
|
it at mesh startup over this server's TLS channel — the web-PKI bootstrap — instead of embedding
|
||
|
it in the client build. A public key is not secret, so this is unauthenticated, mirroring the
|
||
|
JWKS endpoint. [keyId] is a stable, deterministic id (SHA-256 of the raw key, truncated) so the
|
||
|
client can pin the anchor and, later, distinguish a rotated key without guessing."""
|
||
|
if _gw_conf is None or not _gw_conf.mesh_ca_public_key:
|
||
|
abort(NOT_FOUND, "mesh CA is not configured on this server")
|
||
|
pub_b64 = _gw_conf.mesh_ca_public_key
|
||
|
try:
|
||
|
key_id = hashlib.sha256(base64.b64decode(pub_b64)).hexdigest()[:16]
|
||
|
except Exception:
|
||
|
abort(SERVICE_UNAVAILABLE, "mesh CA public key is misconfigured")
|
||
|
return resp_ok({"alg": "Ed25519", "meshCaPublicKey": pub_b64, "keyId": key_id})
|
||
|
|
||
|
|
||
|
@mesh_bp.route("/<string:m_id>/device/<string:d_id>/credential", methods=["GET"])
|
||
|
@require_login
|
||
|
def mesh_device_credential(m_id, d_id): # A1 pull
|
||
| app/test/component/features/mesh_gateway.feature | ||
|---|---|---|
|
@gateway @error_handling
|
||
|
Scenario: Link status for a mesh that does not exist is not found
|
||
|
When I request the link status for a random unknown mesh
|
||
|
Then the response status is 404
|
||
|
Then the response status is 404
|
||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
# Trust-anchor publish — mesh CA public key (fetch + pin at mesh startup)
|
||
|
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
@gateway @happy_path
|
||
|
Scenario: Publish the mesh CA public key so a device can fetch and pin it
|
||
|
When I request the mesh CA public key
|
||
|
Then the response status is 200
|
||
|
And the response publishes the configured Ed25519 mesh CA public key
|
||
| app/test/component/steps/mesh_gateway_steps.py | ||
|---|---|---|
|
body = context["response"].json()
|
||
|
assert "members" in body and isinstance(body["members"], list), (
|
||
|
f"Expected a 'members' list in response: {body}"
|
||
|
)
|
||
|
)
|
||
|
# ── Trust-anchor publish — GET /api/v1/mesh/ca ────────────────────────────────
|
||
|
@when("I request the mesh CA public key")
|
||
|
def when_get_mesh_ca(context: dict, api_client: TomeApiClient):
|
||
|
context["response"] = api_client.get("/api/v1/mesh/ca")
|
||
|
@then("the response publishes the configured Ed25519 mesh CA public key")
|
||
|
def then_mesh_ca_published(context: dict):
|
||
|
body = context["response"].json()
|
||
|
assert body["alg"] == "Ed25519", body
|
||
|
# Must equal mesh_ca_public_key in tome-test-config.yaml (the deterministic test CA),
|
||
|
# i.e. the SAME anchor the gateway verifies leader credentials against.
|
||
|
assert body["meshCaPublicKey"] == "A6EHv/POEL4dcN0Y50vAmWfk1jCbpQ1fHdyGZBJVMbg=", body
|
||
|
assert isinstance(body.get("keyId"), str) and body["keyId"], body
|
||