%%%------------------------------------------------------------------- %%% @doc Macula SDK — Public API for mesh applications. %%% %%% This is the main entry point for applications using the Macula SDK. %%% %%% Apps connect via `connect/2', which returns a `macula_client' %%% pool that internally wraps N peering links to N stations. %%% `publish/4,5', `subscribe/4,5', `unsubscribe/2', `call/5,6', `providers/3,4', %%% `advertise/5', `unadvertise/3', `call_stream/5', %%% `advertise_stream/5', and `unadvertise_stream/3' route through %%% the pool with realm-per-call semantics. See `macula_pubsub' for %%% the slice module of the publish/subscribe surface. %%% %%% LOCAL streaming (`call_stream/2,3', `open_stream/3,4', %%% `advertise_stream/2,3', `unadvertise_stream/1') dispatches %%% in-process via `macula_stream_local' — for unit tests and %%% same-BEAM pairs. %%% %%% Erlang distribution over the mesh ships via `join_mesh/1' (V2 %%% pool carrier) or `join_dist_relay/1' (dedicated dist relay). See %%% `macula_dist_pool' / `macula_dist_relay_client'. %%% %%% @end %%%------------------------------------------------------------------- -module(macula). -include_lib("kernel/include/logger.hrl"). %% Connection -export([connect/2, close/1, child_spec/3, status/1, links/1]). %% Reading fields of peer-supplied maps (D26) -export([field/2, field/3, text/1]). %% Pub/Sub — realm-per-call against a V2 pool -export([subscribe/4, subscribe/5, subscribe_callback/4, unsubscribe/2, publish/4, publish/5]). %% RPC — realm-per-call against a V2 pool -export([call/5, call/6, providers/3, providers/4, call_station/7, call_station/8, advertise/5, unadvertise/3]). %% The pool's own D25 provider authorization, resolved from the DHT and %% verified against the realm key the pool pins — the `authorization' %% opt a direct-dial record publish needs. -export([provider_authorization/3, provider_authorization/4]). %% The pool renews an advertised chain through this (D32); internal. -export([renew_authorization/4, advertise_confidentiality/1, call_seal/5]). %% Signed DHT records — realm-agnostic infrastructure procedures %% (`_dht.put_record', `_dht.find_record', `_dht.find_records_by_type', %% `_dht.records..stored'). The all-zeros realm is the SDK %% convention for protocol-internal traffic. -export([put_record/2, find_record/2, find_record/3, find_records/2, find_records/3, find_records_by_type/2, subscribe_records/3, unsubscribe_records/2]). -export([sign_node_record/2, sign_node_record/3, sign_domain_record/2, withdraw_node_record/3, parse_stations/1]). %% Node-served content (D27): the node that shares content keeps it and %% serves it; stations only relay. A content id (MCID) is 50 bytes: the hash %% tag (2, SHA-384), the codec (16#55 raw, 16#56 manifest), then the 48-byte %% hash. -export([share_content/3, share_content/4, unshare_content/3, get_content/3, get_content/4]). %% A pool link to one station, and the overlay frames a consumer drives it %% with (macula-realm's peer overlay). -export([ensure_station_link/4, overlay_subscribe/3, overlay_unsubscribe/2, send_overlay_frame/2, send_overlay_frame/3]). %% Streaming RPC (LOCAL in-process + V2 pool, see PLAN_MACULA_STREAMING.md) -export([ call_stream/2, call_stream/3, call_stream/5, call_stream_station/7, open_stream/3, open_stream/4, advertise_stream/2, advertise_stream/3, advertise_stream/5, advertise_stream/6, unadvertise_stream/1, unadvertise_stream/3, send/2, send/3, recv/1, recv/2, stream_report/1, close_stream/1, close_send/1, await_reply/1, await_reply/2, set_reply/2, abort/3 ]). %% Cluster (LAN) -export([ensure_distributed/0, monitor_nodes/0, unmonitor_nodes/0]). %% Mesh Distribution -export([join_mesh/1, join_dist_relay/1, dist_relay_client/0]). -ifdef(TEST). -export([join_pool_args/1, renewal/3]). -endif. %% Types -export_type([pool/0, realm/0, provider_io/0, topic/0, procedure/0, stream/0, stream_mode/0, stream_handler/0, m_record/0, record_type/0, record_key/0, mcid/0]). -type pool() :: macula_client:pool(). -type realm() :: <<_:256>>. %% 32-byte realm tag. -type topic() :: binary(). -type procedure() :: binary(). %% The I/O a provider-advertisement resolution runs on, overrideable per %% call in `advertise/5''s and `provider_authorization/4''s `Opts' (the %% same seam discipline as `macula_direct_dial:dial_io/0'): each entry %% defaults to the facade's own function. Tests inject stubs here rather %% than replacing a module other processes call. -type provider_io() :: #{ status => fun((pool()) -> {ok, macula_client:status()} | {error, term()}), find_record => fun((pool(), record_key()) -> {ok, m_record()} | {error, term()}), sign_node_record => fun((pool(), m_record(), #{not_after := integer()}) -> {ok, m_record()} | {error, term()}), realm_key => fun((pool(), realm()) -> {ok, binary()} | none) }. -type stream() :: pid(). -type stream_mode() :: server_stream | client_stream | bidi. -type stream_handler() :: fun((stream(), term()) -> any()). -type m_record() :: macula_record:m_record(). -type record_type() :: macula_record:type_tag(). -type record_key() :: <<_:256>>. %% DHT storage key — `macula_record:storage_key/1' output. %%%=================================================================== %%% Connection — V2 (pool, since 3.11.0) %%%=================================================================== %% @doc Connect to the Macula relay mesh and return a pool handle. %% %% `Seeds' is a list of station endpoints: `#{host, port, expected_node_id}' %% maps, or URL binaries or strings. Every seed names the node_id it %% expects, in the seed or in the `expected_node_id' option; otherwise no %% pool starts and `{error, {seeds, expected_node_id_required}}' is %% returned. The pool spawns one peering link per seed and routes ops with %% replication, replay, and event dedup. Returns immediately; link %% handshakes complete asynchronously. %% %% Honored opts (full reference: `macula_client:opts()'): %% %% %% Legacy opts silently dropped (with a one-shot `logger:notice'): %% `relays' (use the `Seeds' positional argument), `realm' (V2 is %% realm-per-call), `site' (no V2 analog), `connections' (one link %% per seed; add more seeds to grow the pool). %% %% See `macula_client' for the canonical pool implementation and %% `macula_pubsub' for the slice module. -spec connect([macula_client:seed()], macula_client:opts()) -> {ok, pool()} | {error, term()}. connect(Seeds, Opts) when is_list(Seeds), is_map(Opts) -> refused(seeds_checked(Seeds, Opts), fun() -> macula_client:connect(Seeds, Opts) end). %% @doc Stop a V2 pool. Every subscriber receives a final %% `{macula_event_gone, SubRef, pool_closed}' message. -spec close(pool()) -> ok. close(Pool) when is_pid(Pool) -> macula_client:close(Pool). %% @doc OTP child spec to drop a V2 pool into a caller's supervision %% tree. Give `node_identity' as a loader `{Module, Function, Args}' that %% returns `{ok, Key}'. Its `Args' say where the key is and never hold the %% key, because a supervisor logs them when a start fails. See %% `macula_client:child_spec/3'. -spec child_spec(term(), [macula_client:seed()], macula_client:opts()) -> supervisor:child_spec(). child_spec(Id, Seeds, Opts) -> macula_client:child_spec(Id, Seeds, Opts). %% @doc Aggregate health snapshot of a V2 pool. Suitable for %% `/health' or `/status' endpoints; not for hot-loop polling. See %% `macula_client:status/1' for the full shape. -spec status(pool()) -> {ok, macula_client:status()}. status(Pool) when is_pid(Pool) -> macula_client:status(Pool). %% @doc Per-link snapshot of a V2 pool — one entry per spawned link %% with its peer station `node_id' (pubkey), dial `host', `pid', and %% `connected' flag. Use this to resolve a specific station (by pubkey %% or hostname) to its link for targeted, per-station operations. See %% `macula_client:links/1' for the `link_info()' shape. -spec links(pool()) -> {ok, [macula_client:link_info()]}. links(Pool) when is_pid(Pool) -> macula_client:links(Pool). %%%=================================================================== %%% Pub/Sub — realm-per-call against a V2 pool %%%=================================================================== %% @doc Publish to `(Realm, Topic)' on `Pool'. Equivalent to %% `publish/5' with empty opts. -spec publish(pool(), realm(), topic(), term()) -> ok | {error, term()}. publish(Pool, Realm, Topic, Payload) -> macula_pubsub:publish(Pool, Realm, Topic, Payload). %% @doc Publish to `(Realm, Topic)' on `Pool' with options. See %% `macula_pubsub:publish/5' for honored opts. -spec publish(pool(), realm(), topic(), term(), map()) -> ok | {error, term()}. publish(Pool, Realm, Topic, Payload, Opts) -> macula_pubsub:publish(Pool, Realm, Topic, Payload, Opts). %% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool'. %% Equivalent to `subscribe/5' with empty opts. -spec subscribe(pool(), realm(), topic(), pid()) -> {ok, reference()} | {error, {text_too_long | invalid_text, topic}}. subscribe(Pool, Realm, Topic, Subscriber) -> macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber). %% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool' with %% options. The `delivery' option chooses how a single publisher's %% out-of-order arrivals are handled: %% %% Ordering state is kept per publisher, and apart for EVENTs whose %% publisher signature did not verify. %% See `macula_pubsub:subscribe/5'. -spec subscribe(pool(), realm(), topic(), pid(), map()) -> {ok, reference()} | {error, {text_too_long | invalid_text, topic}}. subscribe(Pool, Realm, Topic, Subscriber, Opts) -> macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber, Opts). %% @doc Subscribe with a callback function. The SDK spawns a small %% receiver process internally and invokes the callback once per %% inbound event. See `macula_pubsub:subscribe_callback/4'. -spec subscribe_callback(pool(), realm(), topic(), macula_pubsub:callback()) -> {ok, reference()} | {error, term()}. subscribe_callback(Pool, Realm, Topic, Callback) -> macula_pubsub:subscribe_callback(Pool, Realm, Topic, Callback). %% @doc Drop a pool subscription. Idempotent. -spec unsubscribe(pool(), reference()) -> ok. unsubscribe(Pool, SubRef) when is_pid(Pool), is_reference(SubRef) -> macula_pubsub:unsubscribe(Pool, SubRef). %%%=================================================================== %%% RPC — realm-per-call against a V2 pool %%%=================================================================== %% @doc Call `Procedure' in `Realm' at the provider that serves it: resolve %% the procedure's verified advertisements, authorized against the realm key %% the pool pinned for `Realm', reach the station a candidate names %% directly, and call its provider there. See %% `macula_direct_dial:call/5'. A procedure the pool's linked stations serve %% themselves, such as `_dht.*', goes through %% `macula_client:call_linked_station/5'. -spec call(pool(), realm(), procedure(), term(), 1..600_000) -> {ok, term()} | {error, term()}. call(Pool, Realm, Procedure, Payload, TimeoutMs) -> macula_direct_dial:call(Pool, Realm, Procedure, Payload, TimeoutMs). %% @doc As `call/5', with `Opts'. `#{provider => NodeId}' calls THAT %% provider of `Procedure' and no other: the procedure is resolved and its %% advertisements trust-checked exactly as `call/5' does, and only the named %% provider's are tried. A provider with no trusted advertisement by the %% deadline is `{error, {unresolved, provider_not_advertised}}'. To have %% every provider of a procedure answer, list them with `providers/3,4' and %% make one call per provider. See `macula_direct_dial:call/6'. %% %% The call is sealed to the KEM key of the advertisement it resolves, when %% that advertisement names one (E2E design §8.1). `#{confidential => %% required}' fails the call closed when none does. A lookup never downgrades %% a call, so `confidential => off' is refused as `{error, {confidentiality, %% off_needs_explicit_target}}': a clear call is `call_station/8''s, to a %% target the application names itself. %% %% `report => true' asks for the call's seal report (DESIGN_E2E_SEAL_REPORT): a %% result then comes back as `{ok, Result, Report}', `Report' saying whether the %% request that produced it was sealed and to which key (`sealed' 1 with %% `seal_key_id', or 0), and the `provider' it was addressed to. An error comes %% back as it is. Any value but a boolean is `{error, {invalid_option, report}}'. -spec call(pool(), realm(), procedure(), term(), 1..600_000, #{provider => <<_:256>>, confidential => preferred | required, report => boolean()}) -> {ok, term()} | {ok, term(), macula_station_link:report()} | {error, term()}. call(Pool, Realm, Procedure, Payload, TimeoutMs, Opts) when is_map(Opts) -> macula_direct_dial:call(Pool, Realm, Procedure, Payload, TimeoutMs, Opts). %% @doc As `providers/4', within 5 seconds. -spec providers(pool(), realm(), procedure()) -> {ok, [#{provider := <<_:256>>, station := <<_:256>>}]} | {error, term()}. providers(Pool, Realm, Procedure) -> providers(Pool, Realm, Procedure, 5_000). %% @doc Who provides `Procedure' in `Realm': each provider whose %% advertisement passes the same trust check `call/5' applies, with the %% station it serves from, in one DHT lookup bounded by `TimeoutMs'. A %% provider whose record has not replicated yet is not listed; ask again %% for a fresher answer. See `macula_direct_dial:providers/4'. -spec providers(pool(), realm(), procedure(), 1..600_000) -> {ok, [#{provider := <<_:256>>, station := <<_:256>>}]} | {error, term()}. providers(Pool, Realm, Procedure, TimeoutMs) -> macula_direct_dial:providers(Pool, Realm, Procedure, TimeoutMs). %% @doc Issue a CALL to `Target', a provider's node_id, at ONE specific %% station, dialing it directly even if it is not in the pool's seed set. %% `Station' is a seed URL (e.g. `<<"quic://[::1]:4433">>'). The pool reuses %% an existing link or dials and monitors a new one, waits for the %% handshake, and calls there. This is the direct-dial data path: resolve a %% provider's serving_station and its endpoint, then reach it in one hop. %% See `macula_client:call_station/11'. With no options it carries no signed state to decide on sealing from, so it is %% refused as `{error, {confidentiality, no_signed_state}}' (macula 13.0.0): use `call_station/8' with %% `advertisement', `confidential => required' or `confidential => off' (see `call_seal/5'). -spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), 1..600_000) -> {ok, term()} | {error, term()}. call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs) -> call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, #{}). %% @doc As `call_station/7', presenting a capability token to a gated %% provider via `Opts' (`#{ucan_token => Token}'). Empty/absent = none. %% Slice 7b dual-trust. `Opts' also carries the station this dial must %% prove, `expected_node_id' (see `macula_client:call_station/11'), and may %% set `dial_timeout_ms', how much of `TimeoutMs' the wait for a fresh %% link's handshake may take (default: all of it). `pin_tls_cert => true' %% is REFUSED with %% `{error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}}', %% and `verify' in any value with %% `{error, {refused, {verify, one_verification_mode}}}' (see %% `trust_options_checked/1'). %% %% `report => true' asks for the call's seal report, as `call/6' does: a result then comes back as %% `{ok, Result, Report}' (see `macula_station_link:call/9'). It is honoured here because the pool's own direct dial %% calls through this function. Any value but a boolean is `{error, {invalid_option, report}}', before anything is %% sent. -spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), 1..600_000, map()) -> {ok, term()} | {ok, term(), macula_station_link:report()} | {error, term()}. call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) -> refused(first_error([target_checked(Station, Opts), report_checked(Opts)]), fun() -> do_call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) end). %% `ok', or the refusal of a `report' that is not a boolean, before anything is sent. report_checked(Opts) -> report_valid(macula_direct_dial:report_option(Opts)). report_valid({ok, _Report}) -> ok; report_valid({error, _} = Refused) -> Refused. do_call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) -> Ucan = maps:get(ucan_token, Opts, <<>>), LinkOpts = maps:with([expected_node_id], Opts), DialTimeoutMs = maps:get(dial_timeout_ms, Opts, TimeoutMs), Resolve = fun() -> find_records(Pool, macula_record:procedure_key(Realm, Procedure)) end, sealed_or_refused(call_seal(Target, Realm, Procedure, Opts, Resolve), fun(Seal) -> macula_client:call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Ucan, LinkOpts, DialTimeoutMs, Seal, maps:get(report, Opts, false)) end). sealed_or_refused({ok, Seal}, Call) -> Call(Seal); sealed_or_refused({error, _} = Refused, _Call) -> Refused. %% @private %% @doc How a call to the explicit `Target' is sent (E2E design §8.1, Amendment A1), decided from signed state only, so %% a lookup can deny the call but never downgrade it: %% %% - `advertisement', a verified procedure advertisement of `Target' for this realm and procedure (as `find_records/2' %% or `providers/3' hand one over): sealed to its KEM key when it names one, in the clear when it names none, unless %% `confidential => required', which then fails with `no_kem_key'; %% - `confidential => off': in the clear, the application's own decision; %% - `confidential => required' without an advertisement: `Resolve()' looks up the procedure's advertisements, and %% only one `Target' signed is used; with none, or one naming no key, the call fails with `no_kem_key'; %% - none of these: `{error, {confidentiality, no_signed_state}}'. -spec call_seal(<<_:256>>, realm(), procedure(), map(), fun(() -> {ok, [map()]} | {error, term()})) -> {ok, clear | {sealed_to, binary()}} | {error, {confidentiality, atom()}}. call_seal(_Target, _Realm, _Procedure, #{confidential := off}, _Resolve) -> {ok, clear}; call_seal(Target, Realm, Procedure, #{advertisement := Ad} = Opts, _Resolve) -> sealed_by(advertised_key(Ad, Target, Realm, Procedure), maps:get(confidential, Opts, preferred)); call_seal(Target, Realm, Procedure, #{confidential := required}, Resolve) -> sealed_by(resolved_key(Resolve(), Target, Realm, Procedure), required); call_seal(_Target, _Realm, _Procedure, _Opts, _Resolve) -> {error, {confidentiality, no_signed_state}}. %% The KEM key an advertisement of `Target' for this realm and procedure names, `none' when it names none. advertised_key(Ad, Target, Realm, Procedure) -> advertised_by(macula_record:read_procedure_advertisement(Ad), Target, Realm, Procedure). advertised_by(#{advertiser_node := Target, realm_id := Realm, procedure := Procedure} = Read, Target, Realm, Procedure) -> {ok, maps:get(kem_key, Read, none)}; advertised_by(#{advertiser_node := Target, realm_id := Realm}, Target, Realm, _Procedure) -> {error, {confidentiality, not_the_procedure}}; advertised_by(#{advertiser_node := Target}, Target, _Realm, _Procedure) -> {error, {confidentiality, not_the_realm}}; advertised_by(_AnotherNodes, _Target, _Realm, _Procedure) -> {error, {confidentiality, not_the_target}}. %% The key an advertisement `Target' signed names, among those resolved: a keyed one wins over a stale keyless one %% listed before it; anything else seals nothing. resolved_key({ok, Ads}, Target, Realm, Procedure) -> first_key([advertised_key(Ad, Target, Realm, Procedure) || Ad <- Ads, macula_record:type(Ad) =:= macula_record:type_procedure_advertisement()]); resolved_key(_NothingResolved, _Target, _Realm, _Procedure) -> {ok, none}. first_key(Found) -> keyed_first([KemKey || {ok, KemKey} <- Found, is_binary(KemKey)]). keyed_first([KemKey | _]) -> {ok, KemKey}; keyed_first([]) -> {ok, none}. sealed_by({ok, KemKey}, _Mode) when is_binary(KemKey) -> {ok, {sealed_to, KemKey}}; sealed_by({ok, none}, required) -> {error, {confidentiality, no_kem_key}}; sealed_by({ok, none}, _Preferred) -> {ok, clear}; sealed_by({error, _} = Refused, _Mode) -> Refused. %% @doc Register a procedure handler on a V2 pool and advertise it: %% the pool resolves its own D25 provider authorization — the %% realm-signed `org_directory' and the org-signed %% `procedure_delegation' that names the pool's node id, both fetched %% from the DHT — verifies it against the pinned realm key, and hands %% every link the advertisement spec: for each link the pool signs an %% advertisement naming the station that link is connected to as %% `serving_station', sent as an ADVERTISE frame (signed again on %% reconnect and on link respawn, never past the chain's earlier %% expiry). The pool renews the chain at a third of its remaining life %% (`macula_client:advertise/8', D32): a delegation lives 30 minutes, so %% a provider stays advertised while the realm keeps reissuing, and %% drops out once it stops. The frame registers the procedure at that station; it %% does not put a record in the DHT, which is what a direct-dialling %% caller resolves (`macula_response:advertise_direct/6,7' publishes %% that one). The %% procedure must carry an org namespace, and the pool must run a %% provisioned identity whose delegation the org has published, and %% pin the realm's key (`realm_trust' at connect); a missing piece %% fails fast with `{error, {provider_authorization, _}}'. %% %% `Opts' `auth' sets the procedure's policy: `open' (default, serve %% any identified caller), `{ucan_required, IssuerNodeId}' (gated to %% tokens from one known node), or `{realm_member_required, RealmKeyId, %% RequiredCan}' (gated to realm membership at a specific tier) -- see %% `macula_client:auth_policy()' for the full set. %% %% `stations => [StationNodeId]' registers the procedure on the links to %% those stations only, each named by the node_id its link pins, and a link %% respawned later registers it again only when its station is one of them. %% A station the pool holds no link to is refused as `{error, %% {station_not_linked, StationNodeId}}', an empty list as `{error, %% {stations, empty}}' and anything but a list of 32-byte node ids as %% `{error, {stations, malformed}}'; nothing is registered or kept. Without %% it every link registers the procedure. %% %% `Opts' may also carry `advertise', an arity-6 override for the pool %% fan-out, which then decides where the procedure registers (see %% `macula_response:advertise_opts()'), and the `provider_io/0' seam %% entries the resolution reads its DHT calls from. -spec advertise(pool(), realm(), procedure(), macula_client:handler(), map()) -> ok | {error, term()}. advertise(Pool, Realm, Procedure, Handler, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32 -> on_stations(stations_opt(Opts), fun(Stations) -> Policy = maps:get(auth, Opts, open), Advertise = maps:get(advertise, Opts, fun(P, R, Pr, H, Po, Spec) -> macula_client:advertise(P, R, Pr, H, Po, Spec, Stations, renewal(R, Pr, Opts)) end), advertise_authorized(Pool, Realm, Procedure, Opts, fun(Spec) -> Advertise(Pool, Realm, Procedure, Handler, Policy, Spec) end) end). %% The `stations' an advertisement names, checked before anything is resolved %% or registered: `all' when it names none. stations_opt(#{stations := []}) -> {error, {stations, empty}}; stations_opt(#{stations := Stations}) -> node_ids(Stations, Stations); stations_opt(#{}) -> {ok, all}. node_ids([], Stations) -> {ok, Stations}; node_ids([<<_:256>> | Rest], Stations) -> node_ids(Rest, Stations); node_ids(_NotANodeIdList, _Stations) -> {error, {stations, malformed}}. on_stations({ok, Stations}, Fun) -> Fun(Stations); on_stations({error, _} = Refused, _Fun) -> Refused. %% @doc Stop advertising a procedure on a V2 pool. -spec unadvertise(pool(), realm(), procedure()) -> ok. unadvertise(Pool, Realm, Procedure) -> macula_client:unadvertise(Pool, Realm, Procedure). %%%=================================================================== %%% Signed DHT records (v3.3.0) %%%=================================================================== %%% %%% Records are typed payloads signed in the signed-object format of %%% DESIGN_PQ_SIGNED_FRAMES_AND_RECORDS.md and stored in the relay %%% mesh's distributed hash table. A record travels as its wire form: %%% stations store and forward its bytes as received. A record reaches %%% a caller only after it verifies under the node's crypto profile %%% (`macula_record:verify/2'): `find_record/2' returns the refusal of %%% one that does not, and the list functions and `subscribe_records/3' %%% drop it. %%% %%% See `macula_record' for the record shape, the typed %%% constructors (`node_record/3', `realm_directory/3', %%% `realm_stations/2', `procedure_advertisement/4', %%% `content_announcement/3', `tombstone/2', and the foundation_* %%% constructors), and `storage_key/1' for the DHT addressing rule. %%% %%% Two complementary retrieval paths: %%% %%% - `find_record/2' — fetch one record by its %%% `storage_key/1' output %%% - `find_records_by_type/2' — list every record of a given %%% type tag %%% %%% Plus a live-update channel: %%% %%% - `subscribe_records/3' — receive new records of a type %%% as they are stored %% Procedure + topic shape — hidden from API consumers but exposed %% as documentation. The relay backend (hecate-station and successors) %% MUST advertise these procedures and publish on the per-type %% record-stored topic for the SDK to function. DHT traffic travels %% under the all-zeros realm tag (protocol-internal infrastructure; %% the same convention `macula_dist_pool' uses for tunnel frames). -define(DHT_REALM, <<0:256>>). -define(DHT_PUT_RECORD_PROC, <<"_dht.put_record">>). -define(DHT_FIND_RECORD_PROC, <<"_dht.find_record">>). -define(DHT_FIND_RECORDS_PROC, <<"_dht.find_records">>). -define(DHT_FIND_RECORDS_BY_TYPE_PROC, <<"_dht.find_records_by_type">>). -define(DHT_RECORD_TIMEOUT_MS, 5_000). %% @doc Store a signed record in the mesh DHT via a V2 pool. %% %% Build the record via the typed constructors in `macula_record' %% (`node_record/3,4', `content_announcement/3,4', `tombstone/2,3', %% `realm_directory/3,4', `procedure_advertisement/4,5', etc.) and sign %% it. A record this node signs about itself is signed by its pool, %% which holds the node identity key: through %% `macula_client:sign_node_record/2', a domain record (tags 0x20 to %% 0xFF) through `macula_client:sign_domain_record/2', and the %% tombstone of either through `macula_client:withdraw_node_record/3'. %% A realm-, org- or %% foundation-signed record is signed with `macula_record:sign/2' and %% its signer's key. Pass the signed record or its wire form %% (`macula_record:encode/1'). The record travels as its %% wire form; the station verifies it on receipt and stores it under %% `macula_record:storage_key/1', propagating to the K-nearest peers. -spec put_record(pool(), m_record() | binary()) -> ok | {error, term()}. put_record(Pool, #{key := _, tbs := _, signature := _} = Signed) when is_pid(Pool) -> put_record(Pool, macula_record:encode(Signed)); put_record(Pool, Wire) when is_pid(Pool), is_binary(Wire) -> %% Record bytes paced per pool, so a bulk writer stays under the %% station's STORE allowance (D28, 3.5) instead of running into stored 0. %% One bucket per pool is conservative for a pool of several links: its %% calls fan out one link at a time, and the per-link pacer in %% macula_station_link:put_record/3 paces a caller that writes straight %% through a link. ok = macula_store_pacer:await(Pool, byte_size(Wire)), classify_put(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_PUT_RECORD_PROC, Wire, ?DHT_RECORD_TIMEOUT_MS)). classify_put({ok, ok}) -> ok; %% The reply's atom crossed the wire as text and decoded back to the %% codec's {text, _} marker (D26 makes no atom on the wire). classify_put({ok, {text, <<"ok">>}}) -> ok; classify_put({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_put({error, _} = E) -> E. %% @doc Fetch a record from the mesh DHT by its %% `macula_record:storage_key/1'. %% %% Returns `{error, not_found}' when no record exists at the key. The %% record is verified under the node's crypto profile before it is %% returned; one that does not verify returns its refusal from %% `macula_record:verify/2', such as `{error, expired}'. -spec find_record(pool(), record_key()) -> {ok, m_record()} | {error, not_found | term()}. find_record(Pool, Key) -> find_record(Pool, Key, ?DHT_RECORD_TIMEOUT_MS). %% @doc As `find_record/2', waiting at most `TimeoutMs' for the reply, for a %% caller that bounds its work by a deadline of its own. -spec find_record(pool(), record_key(), pos_integer()) -> {ok, m_record()} | {error, not_found | term()}. find_record(Pool, Key, TimeoutMs) when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32, is_integer(TimeoutMs), TimeoutMs > 0 -> classify_find(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORD_PROC, #{key => Key}, TimeoutMs)). classify_find({ok, Wire}) when is_binary(Wire) -> with_profile(fun(Profile) -> macula_record:verify(Wire, Profile) end); classify_find({ok, not_found}) -> {error, not_found}; %% The reply's atom crossed the wire as text and decoded back to the %% codec's {text, _} marker (D26 makes no atom on the wire). classify_find({ok, {text, <<"not_found">>}}) -> {error, not_found}; classify_find({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_find({error, _} = E) -> E. %% @doc Fetch EVERY record stored at `Key' — the full multi-value %% set, e.g. every `procedure_advertisement' under one procedure's %% storage key. Where `find_record/2' returns the first record (or %% `not_found'), this returns the whole list, empty when none. %% %% The relay's local store is a signer-deduped multiset: one record %% per signing key at a storage key, so N providers of one procedure %% return N records. Each record is verified under the node's crypto %% profile, and a record that does not verify is dropped. -spec find_records(pool(), record_key()) -> {ok, [m_record()]} | {error, term()}. find_records(Pool, Key) -> find_records(Pool, Key, ?DHT_RECORD_TIMEOUT_MS). %% @doc As `find_records/2', waiting at most `TimeoutMs' for the reply, for a %% caller that bounds its work by a deadline of its own. -spec find_records(pool(), record_key(), pos_integer()) -> {ok, [m_record()]} | {error, term()}. find_records(Pool, Key, TimeoutMs) when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32, is_integer(TimeoutMs), TimeoutMs > 0 -> classify_find_list(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORDS_PROC, #{key => Key}, TimeoutMs)). classify_find_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires); classify_find_list({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_find_list({error, _} = E) -> E. %% @doc Return every record of a given type currently visible from %% the pool's connected stations, each verified under the node's %% crypto profile; a record that does not verify is dropped. %% %% Coverage depends on each station's view of the DHT — a single %% station sees its local replicas plus whatever its peers have %% gossiped. Aggregating across the full mesh requires querying %% multiple stations and deduplicating by record key. -spec find_records_by_type(pool(), record_type()) -> {ok, [m_record()]} | {error, term()}. find_records_by_type(Pool, Type) when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255 -> classify_list(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORDS_BY_TYPE_PROC, #{type => Type}, ?DHT_RECORD_TIMEOUT_MS)). classify_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires); classify_list({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_list({error, _} = E) -> E. %% The records among Wires that verify under the node's crypto profile; %% the rest are dropped. verified_records(Wires) -> with_profile(fun(Profile) -> {ok, [Record || Wire <- Wires, {ok, Record} <- [macula_record:verify(Wire, Profile)]]} end). %% Verify with the node's crypto profile, or return the refusal of a %% node that has none. with_profile(Verify) -> profile_verified(macula_crypto_profile:configured(), Verify). profile_verified({ok, Profile}, Verify) -> Verify(Profile); profile_verified({error, _} = Refusal, _Verify) -> Refusal. %% @doc Subscribe to live record-stored events filtered by type. %% %% The callback receives each newly-stored record of the given type %% that verifies under the node's crypto profile. Returns a %% subscription reference for `unsubscribe_records/2'. %% Topic shape is `_dht.records..stored', rendered with the %% type tag as a decimal integer for log friendliness. -spec subscribe_records(pool(), record_type(), fun((m_record()) -> any())) -> {ok, reference()} | {error, term()}. subscribe_records(Pool, Type, Callback) when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255, is_function(Callback, 1) -> Topic = record_stored_topic(Type), macula_pubsub:subscribe_callback(Pool, ?DHT_REALM, Topic, wrap_record_callback(Callback)). %% @doc Cancel a `subscribe_records/3' subscription. -spec unsubscribe_records(pool(), reference()) -> ok. unsubscribe_records(Pool, Ref) when is_pid(Pool), is_reference(Ref) -> macula_pubsub:unsubscribe(Pool, Ref). %%%=================================================================== %%% Signing records this node signs about itself — the pool holds the key %%%=================================================================== %% @doc Sign a record this node signs about itself with the pool's node %% identity key, in the pool's own process: a node record, a procedure %% advertisement or a content announcement that names this node. See %% `macula_client:sign_node_record/2' for the record checks and %% refusals. `Opts' `not_after' bounds the signed record's expiry: the %% pool judges it on its own clock, refuses a bound already passed as %% `{error, not_after_passed}', ends the record at the bound when the %% bound comes before the record's lifetime runs out, and keeps the %% built lifetime otherwise. -spec sign_node_record(pool(), m_record()) -> {ok, m_record()} | {error, term()}. sign_node_record(Pool, Record) when is_pid(Pool) -> macula_client:sign_node_record(Pool, Record). -spec sign_node_record(pool(), m_record(), #{not_after := integer()}) -> {ok, m_record()} | {error, term()}. sign_node_record(Pool, Record, Opts) when is_pid(Pool), is_map(Opts) -> macula_client:sign_node_record(Pool, Record, Opts). %% @doc Sign a domain record (tags 0x20 to 0xFF) as this node, with the %% pool's node identity key, in the pool's own process. See %% `macula_client:sign_domain_record/2'. -spec sign_domain_record(pool(), m_record()) -> {ok, m_record()} | {error, term()}. sign_domain_record(Pool, Record) when is_pid(Pool) -> macula_client:sign_domain_record(Pool, Record). %% @doc Withdraw a record this node signed — a node record, a procedure %% advertisement, a content announcement or a domain record — with a %% tombstone signed by the pool's node identity key, in the pool's own %% process. See `macula_client:withdraw_node_record/3'. -spec withdraw_node_record(pool(), m_record() | binary(), macula_record:reason()) -> {ok, m_record()} | {error, term()}. withdraw_node_record(Pool, Withdrawn, Reason) when is_pid(Pool) -> macula_client:withdraw_node_record(Pool, Withdrawn, Reason). %% @doc Parse a `MACULA_STATIONS' seed list: comma-separated %% `@:' entries, the node id as 64 lowercase hex %% characters, the host a DNS name, an IPv4 address or a bracketed IPv6 %% address, the port 1 to 65535. Returns the seeds in order, each pinned %% to its node id. A refusal names the entry's position and what is %% wrong with it, and carries neither the value, nor a node id, nor a %% host. See `macula_stations:parse/1'. -spec parse_stations(binary()) -> {ok, [macula_stations:seed()]} | {error, term()}. parse_stations(Value) when is_binary(Value) -> macula_stations:parse(Value). record_stored_topic(Type) -> iolist_to_binary([<<"_dht.records.">>, integer_to_binary(Type), <<".stored">>]). %% Adapt a 1-arg `(Record) -> any()' user callback to the 3-arg %% `(Topic, Payload, Meta) -> any()' shape `macula_pubsub' delivers. %% %% PubSub delivers the payload as the record's wire form (the %% substrate's `record_fanout' publishes `macula_record:encode/1' %% output on the `_dht.records..stored' topic). The record is %% verified here under the node's crypto profile, so the callback %% receives only verified records. A payload that does not verify is %% dropped silently: surfacing it would force every user to handle %% refusals on a protocol-internal channel. wrap_record_callback(Fun) -> fun(_Topic, Payload, _Meta) -> apply_callback_with_verified(Fun, Payload) end. apply_callback_with_verified(Fun, Payload) -> callback_with(with_profile(fun(Profile) -> macula_record:verify(Payload, Profile) end), Fun). callback_with({ok, Record}, Fun) -> Fun(Record), ok; callback_with({error, _}, _Fun) -> ok. %%%=================================================================== %%% Node-served content (D27) %%%=================================================================== -type mcid() :: <<_:400>>. %% @doc Share `Bytes' in `Realm', as `share_content/4' with no options. -spec share_content(pool(), realm(), binary()) -> {ok, mcid()} | {error, term()}. share_content(Pool, Realm, Bytes) -> share_content(Pool, Realm, Bytes, #{}). %% @doc Share `Bytes' in `Realm' from this node (D27): the pool's sharer keeps %% them, serves them on the node's own content procedure and announces their %% root in the DHT, naming the realm, the station the node is reachable %% through and that procedure (`macula_content_sharer'). Returns the root's %% content id: bytes of at most 256 KiB are one raw block, larger bytes a %% manifest over their chunks. The content is available while this node is %% online. `Opts': `name', and `org', the org whose delegation serves the %% content (`/content_v1_'); without `org' the node serves in %% its own namespace (`~/content_v1'). -spec share_content(pool(), realm(), binary(), map()) -> {ok, mcid()} | {error, term()}. share_content(Pool, Realm, Bytes, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Bytes), is_map(Opts) -> shared_by(macula_content_sharer_sup:sharer(Pool), Realm, Bytes, Opts). shared_by({ok, Sharer}, Realm, Bytes, Opts) -> macula_content_sharer:share(Sharer, Realm, Bytes, Opts); shared_by({error, _} = Error, _Realm, _Bytes, _Opts) -> Error. %% @doc Stop sharing the root `MCID' in `Realm': its announcement is withdrawn %% and its bytes dropped. Idempotent. -spec unshare_content(pool(), realm(), mcid()) -> ok. unshare_content(Pool, Realm, MCID) when is_pid(Pool) -> unshared_by(macula_content_sharer_sup:sharer(Pool), Realm, MCID). unshared_by({ok, Sharer}, Realm, MCID) -> macula_content_sharer:unshare(Sharer, Realm, MCID); unshared_by({error, _}, _Realm, _MCID) -> ok. %% @doc Fetch `MCID' in `Realm', as `get_content/4' with no options. -spec get_content(pool(), realm(), mcid()) -> {ok, binary()} | {error, term()}. get_content(Pool, Realm, MCID) -> get_content(Pool, Realm, MCID, #{}). %% @doc Fetch `MCID' from a node that shares it in `Realm' (D27): find its %% announcements, open a stream to a sharer through the station it %% announced, and verify every block and the manifest against their content %% ids (`macula_content_fetch'). A sharer that fails moves the fetch to the %% next; `{error, {unavailable, [{Sharer, Reason}]}}' when all fail, %% `{error, not_shared}' when none announces it, `{error, invalid_mcid}' for %% an id that is not tag 2. `Opts': `max_bytes' (256 MiB), `max_chunks', %% `chunk_timeout_ms', `parallel'. No realm key is needed: content verifies %% itself. -spec get_content(pool(), realm(), mcid(), map()) -> {ok, binary()} | {error, term()}. get_content(Pool, Realm, MCID, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_map(Opts) -> macula_content_fetch:get(Pool, MCID, Opts#{realm => Realm}). %%%=================================================================== %%% Streaming RPC (v1.5.0+) %%%=================================================================== %%% %%% Streaming RPC ships in two phases (see PLAN_MACULA_STREAMING.md): %%% %%% Phase 1 (this release) — LOCAL dispatch only. Client and server %%% halves both live in the same BEAM and are paired in-process. The %%% public surface below is what cross-node streaming will use; only %%% the transport behind the scenes changes in Phase 2. %%% %%% Phase 2 — wire STREAM_OPEN / STREAM_DATA / STREAM_END / STREAM_ERROR %%% / STREAM_REPLY frames through QUIC, one QUIC stream per call_id. %%% %%% Patterns supported (gRPC taxonomy): %%% server_stream — single Args, streamed reply %%% client_stream — streamed Args, single reply %%% bidi — duplex %%% %%% Example (server-stream): %%% %%% ok = macula:advertise_stream(<<"foo.count">>, server_stream, %%% fun(Stream, #{n := N}) -> %%% [ok = macula:send(Stream, integer_to_binary(I)) %%% || I <- lists:seq(1, N)], %%% macula:close_stream(Stream) %%% end), %%% {ok, S} = macula:call_stream(<<"foo.count">>, #{n => 5}), %%% drain(S). %%% %%% drain(S) -> %%% case macula:recv(S) of %%% {chunk, Bin} -> io:format("~s~n", [Bin]), drain(S); %%% eof -> ok %%% end. %% @doc Open a LOCAL in-process server-stream call. Used for unit %% tests and same-BEAM dispatch via `macula_stream_local'. -spec call_stream(procedure(), term()) -> {ok, stream()} | {error, term()}. call_stream(Procedure, Args) when is_binary(Procedure) -> call_stream(Procedure, Args, #{}). %% @doc Open a LOCAL in-process server-stream call with options. -spec call_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream(Procedure, Args, Opts) when is_binary(Procedure), is_map(Opts) -> macula_stream_local:call_stream(Procedure, Args, Opts). %% @doc Open a streaming RPC to `Procedure''s provider in `Realm': resolve %% the provider through its `procedure_advertisement' and open the stream %% at its serving station, naming the provider as the target, as %% `call/5' does for a single-reply call. Same as %% `macula_direct_dial:call_stream/5'. The returned stream is bound to %% that station's link (errors with `peer_down' if the link dies; caller %% re-opens). %% `Opts' `ucan_token' presents a UCAN to a streaming procedure %% advertised with an `auth' policy (see `advertise_stream/6'). %% An open whose signed STREAM_OPEN would be longer than %% `max_stream_open_bytes' (1 MiB by default) returns %% `{error, {open_too_large, Limit}}' without sending anything. %% The stream is sealed as `call/6' seals a call: `confidential => required' %% fails closed without a key, and `confidential => off' is refused as %% `{error, {confidentiality, off_needs_explicit_target}}' (see %% `call_stream_station/7'). -spec call_stream(pool(), realm(), procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream(Pool, Realm, Procedure, Args, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), is_map(Opts) -> macula_direct_dial:call_stream(Pool, Realm, Procedure, Args, Opts). %% @doc Open a streaming RPC to `Target', a provider's node_id, by DIALING %% a specific station directly (direct-dial): the streaming analogue of %% `call_station/7'. Compose it with DHT resolution %% (`find_records' -> `read_procedure_advertisement' -> `station_endpoint') %% to reach a stream provider in one hop, exactly as a unary caller does. %% `Opts' may set `dial_timeout_ms' (default 10_000) and a `mode'. %% `Opts' also names the station this dial must prove, `expected_node_id' %% (see `macula_client:call_station/11'); `pin_tls_cert => true' and `verify' %% are REFUSED as `call_station/8' describes. %% %% Whether the STREAM_OPEN and every later frame is sealed is decided from %% signed state only, as for `call_station/8' (`call_seal/5', E2E design %% §8.1): `advertisement', `confidential => off' or `confidential => %% required'. With none of them the open is refused %% `{error, {confidentiality, no_signed_state}}' before anything is sent. -spec call_stream_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) when is_pid(Pool), is_binary(Target), byte_size(Target) =:= 32, is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), is_map(Opts) -> refused(first_error([target_checked(Station, Opts), report_valid(macula_direct_dial:stream_report_option(Opts))]), fun() -> do_call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) end). %% The pool is handed the seal decided, not the signed state it was decided from. do_call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) -> Resolve = fun() -> find_records(Pool, macula_record:procedure_key(Realm, Procedure)) end, sealed_or_refused(call_seal(Target, Realm, Procedure, Opts, Resolve), fun(Seal) -> macula_client:call_stream_station(Pool, Station, Target, Realm, Procedure, Args, maps:without([advertisement, confidential], Opts), Seal) end). %% @doc A pool link to `Station', for a consumer that must address one %% station through one link, as an overlay does: a live link the pool %% already holds to that station, else one it dials. The station is %% pinned by the `expected_node_id' in its seed map or in `Opts', which a %% link requires (D5, D16); `pin_tls_cert => true' and `verify' are %% refused as `call_station/8' describes. Answers `{ok, Link}' once the %% link's handshake has completed, within `TimeoutMs', or %% `{error, not_connected}'. The pool owns and monitors the link: it %% respawns it and ends it with the pool, so a consumer holds the pid and %% never stops it. Drive it with `overlay_subscribe/3' and %% `send_overlay_frame/2,3'. -spec ensure_station_link(pool(), macula_client:seed(), map(), pos_integer()) -> {ok, pid()} | {error, term()}. ensure_station_link(Pool, Station, Opts, TimeoutMs) when is_pid(Pool), is_map(Opts), is_integer(TimeoutMs), TimeoutMs > 0 -> refused(target_checked(Station, Opts), fun() -> macula_client:ensure_station_link(Pool, Station, Opts, TimeoutMs) end). %% @doc Take the overlay frames for `Realm' arriving on `Link' (from %% `ensure_station_link/4'), delivered to `Subscriber' as %% `{macula_overlay_frame, SubRef, Frame, Meta}', where `Meta' names the %% frame's authenticated `sender' and, for a relayed frame, the station it %% came `via'. A frame from the connected peer, or a GOSSIP, for a realm %% nobody on this link subscribed to is dropped. -spec overlay_subscribe(pid(), realm(), pid()) -> {ok, reference()} | {error, term()}. overlay_subscribe(Link, Realm, Subscriber) -> macula_station_link:overlay_subscribe(Link, Realm, Subscriber). %% @doc Drop an overlay subscription. Idempotent. -spec overlay_unsubscribe(pid(), reference()) -> ok. overlay_unsubscribe(Link, SubRef) -> macula_station_link:overlay_unsubscribe(Link, SubRef). %% @doc Send a pre-built, pre-signed overlay frame to the peer at the other %% end of `Link'. `{error, not_connected}' until its handshake completes. -spec send_overlay_frame(pid(), macula_frame:frame()) -> ok | {error, term()}. send_overlay_frame(Link, Frame) -> macula_station_link:send_overlay_frame(Link, Frame). %% @doc Send a pre-built, pre-signed overlay frame to `TargetPeer' (its %% node_id), relayed by the station `Link' is connected to. %% `{error, not_connected}' until the link's handshake completes. -spec send_overlay_frame(pid(), <<_:256>>, macula_frame:frame()) -> ok | {error, term()}. send_overlay_frame(Link, TargetPeer, Frame) -> macula_station_link:send_overlay_frame(Link, TargetPeer, Frame). %% @doc Open a LOCAL in-process client-stream or bidi call. Used %% for unit tests and same-BEAM dispatch via `macula_stream_local'. -spec open_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}. open_stream(Procedure, Args, Opts) when is_binary(Procedure), is_map(Opts) -> macula_stream_local:open_stream(Procedure, Args, Opts). %% @doc Open a LOCAL in-process stream with explicit mode. -spec open_stream(procedure(), term(), map(), stream_mode()) -> {ok, stream()} | {error, term()}. open_stream(Procedure, Args, Opts, Mode) when is_binary(Procedure), is_map(Opts), is_atom(Mode) -> macula_stream_local:open_stream(Procedure, Args, Opts#{mode => Mode}). %% @doc Advertise a LOCAL in-process streaming procedure %% (default: server_stream). -spec advertise_stream(procedure(), stream_handler()) -> ok | {error, term()}. advertise_stream(Procedure, Handler) when is_binary(Procedure), is_function(Handler, 2) -> advertise_stream(Procedure, server_stream, Handler). %% @doc Advertise a LOCAL in-process streaming procedure with mode. -spec advertise_stream(procedure(), stream_mode(), stream_handler()) -> ok | {error, term()}. advertise_stream(Procedure, Mode, Handler) when is_binary(Procedure), is_atom(Mode), is_function(Handler, 2) -> macula_stream_local:advertise(Procedure, Mode, Handler). %% @doc Register a streaming procedure handler on a V2 pool. Fans out to %% every healthy link and stores in pool state for replay on link %% respawn. A caller reaches this provider only through a %% `procedure_advertisement' record that names it; registering the %% handler publishes none. See `macula_client:advertise_stream/5'. -spec advertise_stream(pool(), realm(), procedure(), stream_mode(), stream_handler()) -> ok | {error, term()}. advertise_stream(Pool, Realm, Procedure, Mode, Handler) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), (Mode =:= server_stream orelse Mode =:= client_stream orelse Mode =:= bidi), is_function(Handler, 2) -> advertise_stream(Pool, Realm, Procedure, Mode, Handler, #{}). %% @doc As `advertise_stream/5', with `Opts'. `auth' sets the streaming %% procedure's policy, the same set `advertise/5' takes: `open' (default), %% `{ucan_required, IssuerNodeId}' or `{realm_member_required, RealmKeyId, %% RequiredCan}'. A consumer presents its token with `call_stream/5''s %% `ucan_token' opt. `stations' registers it on the links to those stations %% only, as `advertise/5' does. -spec advertise_stream(pool(), realm(), procedure(), stream_mode(), stream_handler(), map()) -> ok | {error, term()}. advertise_stream(Pool, Realm, Procedure, Mode, Handler, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), (Mode =:= server_stream orelse Mode =:= client_stream orelse Mode =:= bidi), is_function(Handler, 2), is_map(Opts) -> on_stations(stations_opt(Opts), fun(Stations) -> Policy = maps:get(auth, Opts, open), advertise_authorized(Pool, Realm, Procedure, Opts, fun(Spec) -> macula_client:advertise_stream(Pool, Realm, Procedure, Mode, Handler, Policy, Spec, Stations, renewal(Realm, Procedure, Opts)) end) end). %% The provider-authorization resolution behind `advertise' and %% `advertise_stream': the pool's own D25 chain, fetched from the DHT, %% signed and verified before a single frame goes out. Any missing %% piece fails fast under `provider_authorization'. What reaches the %% pool is the advertisement spec, not the signed check: each link %% signs its own advertisement naming the station it is connected to, %% bounded by the chain (macula_station_link:advertisement_spec()). advertise_authorized(Pool, Realm, Procedure, Opts, Fun) -> confidential(advertise_confidentiality(Opts), fun(Confidentiality) -> signed_provider_advertisement(Pool, Realm, Procedure, provider_io(Opts), fun(Spec) -> Fun(maps:merge(Spec, Confidentiality)) end) end). %% Whether an advertisement names this node's KEM key (E2E design §8.2, Amendment A1), as spec fields: `kem' names %% it, and `confidential => required' also refuses every clear call. A key is named only once the node is switched on %% with the `kem_advertise' application setting, which stays `disabled' until every station runs the release that %% stores a keyed advertisement, and only when `confidential' is not `off'. `required' while switched off is refused: %% the provider would name no key and refuse every clear call, and so be unreachable. %% @private %% @doc What an advertise's `confidential' option and this node's `kem_advertise' switch decide, as the spec fields an %% advertisement carries: `kem' when it names the KEM key, `confidential => required' when it also refuses clear %% calls. The pool's ADVERTISE and a direct-dial DHT record decide the same way (`macula_direct_dial'). -spec advertise_confidentiality(map()) -> {ok, #{kem => true, confidential => required}} | {error, term()}. advertise_confidentiality(Opts) -> confidentiality(maps:get(confidential, Opts, preferred), kem_advertise()). confidentiality(Mode, _Switch) when Mode =/= preferred, Mode =/= required, Mode =/= off -> {error, {confidentiality, {not_a_mode, Mode}}}; confidentiality(required, disabled) -> {error, {confidentiality, kem_advertise_disabled}}; confidentiality(off, _Switch) -> {ok, #{}}; confidentiality(preferred, disabled) -> {ok, #{}}; confidentiality(preferred, enabled) -> {ok, #{kem => true}}; confidentiality(required, enabled) -> {ok, #{kem => true, confidential => required}}. confidential({ok, Confidentiality}, Fun) -> Fun(Confidentiality); confidential({error, _} = Refused, _Fun) -> Refused. %% The node's `kem_advertise' switch: `disabled' unless set to `enabled'. A value outside the two raises, loudly, rather %% than quietly meaning off. kem_advertise() -> kem_switch(application:get_env(macula, kem_advertise, disabled)). kem_switch(enabled) -> enabled; kem_switch(disabled) -> disabled; kem_switch(Other) -> erlang:error({kem_advertise, {not_a_switch, Other}}). %% How the pool renews the chain an advertisement carries (D32): the same %% resolution, over the same `provider_io/0' seams the advertise was given, %% named as an MFA the pool applies with itself prepended. Only the seam %% entries travel, and in production there are none. renewal(Realm, Procedure, Opts) -> {?MODULE, renew_authorization, [Realm, Procedure, maps:with([status, find_record, sign_node_record, realm_key, confidential], Opts)]}. %% @private %% @doc The pool's renewal of an advertised chain (D32, macula#38): the chain %% resolved, signed and verified again as `advertise/5' does, answered as the %% fresh advertisement spec, or the same refusal `advertise/5' would give. A %% `procedure_delegation' lives 30 minutes, so the pool calls this at a third %% of the spec's remaining life, from a worker of its own. Internal: called %% through the MFA `advertise/5' registers. -spec renew_authorization(pool(), realm(), procedure(), map()) -> {ok, macula_station_link:advertisement_spec()} | {error, term()}. renew_authorization(Pool, Realm, Procedure, Seams) -> advertise_authorized(Pool, Realm, Procedure, Seams, fun(Spec) -> {ok, Spec} end). %% @doc Resolve this pool's own D25 provider authorization for an %% org-namespaced `Procedure' under `Realm' — the realm-signed %% `org_directory' and the org-signed `procedure_delegation' naming the %% pool's node id, both fetched from the DHT and verified against the %% realm key the pool pinned for `Realm' at connect — as the %% `authorization' opt `macula_direct_dial:publish_advertisement/5' %% and `macula_response:advertise_direct/6,7' take: a map of the two %% records' encoded wire forms. `advertise/5' resolves this same chain %% itself for the wire frame; a caller that publishes the direct-dial %% record needs it explicitly, since the station refuses an %% org-namespaced record without one (`no_authorization'). %% %% Fails fast with `{error, {provider_authorization, _}}' when the %% procedure has no org namespace, a chain piece is missing from the %% DHT, the pool pinned no key for `Realm', or the chain does not %% verify against that key — the same failures `advertise/5' reports. %% %% A procedure in this node's own namespace, `~/', needs %% no chain (D25 item 6, revised 2026-09-24): the answer is `{ok, undefined}', %% which `publish_advertisement/5' reads as no authorization. Another node's %% namespace is `{error, {provider_authorization, not_own_namespace}}'. -spec provider_authorization(pool(), realm(), procedure()) -> {ok, #{org_directory := binary(), procedure_delegation := binary()} | undefined} | {error, term()}. provider_authorization(Pool, Realm, Procedure) -> provider_authorization(Pool, Realm, Procedure, #{}). %% @doc As `provider_authorization/3', reading the resolution's DHT %% calls from the `provider_io/0' seam entries in `Opts' (defaults to %% the facade's own functions) — the seam `advertise/5' accepts too. -spec provider_authorization(pool(), realm(), procedure(), provider_io()) -> {ok, #{org_directory := binary(), procedure_delegation := binary()} | undefined} | {error, term()}. provider_authorization(Pool, Realm, Procedure, Opts) -> case signed_provider_advertisement(Pool, Realm, Procedure, provider_io(Opts), fun(#{authorization := A}) -> A; (#{}) -> undefined end) of {error, _} = E -> E; Authorization -> {ok, Authorization} end. provider_io(Opts) -> #{status => maps:get(status, Opts, fun status/1), find_record => maps:get(find_record, Opts, fun find_record/2), sign_node_record => maps:get(sign_node_record, Opts, fun sign_node_record/3), realm_key => maps:get(realm_key, Opts, fun macula_client:realm_key/2)}. %% The pool's own D25 chain, resolved from the DHT, signed and verified %% against the realm key the pool pins, passed to `Fun' as the %% advertisement spec: the verified authorization and the bound no %% advertisement carrying it may pass. Any missing piece fails fast %% under `provider_authorization'. signed_provider_advertisement(Pool, Realm, Procedure, Io, Fun) -> case macula_record:procedure_org(Procedure) of none -> {error, {provider_authorization, no_org_namespace}}; {error, malformed} -> {error, {provider_authorization, malformed_procedure}}; {org, <<"~", _/binary>>} -> case (maps:get(status, Io))(Pool) of {ok, #{self_node_id := NodeId}} -> own_namespace_spec(Realm, Procedure, NodeId, Fun); {error, _} = E -> E end; {org, Org} -> case (maps:get(status, Io))(Pool) of {ok, #{self_node_id := NodeId}} -> resolve_org_directory(Io, Pool, Realm, Org, Procedure, NodeId, Fun); {error, _} = E -> E end end. %% A procedure in this node's own namespace, `~/', is %% authorized by the advertisement's own signature (D25 item 6, revised %% 2026-09-24): no org directory, no delegation, no bound but the %% advertisement's own lifetime. What each link signs is checked by the %% same rule every verifier uses, before any spec reaches the pool; another %% node's namespace is refused here. own_namespace_spec(Realm, Procedure, NodeId, Fun) -> own_namespace_checked( macula_record:own_namespace( macula_record:procedure_advertisement(NodeId, Realm, Procedure, NodeId)), Fun). own_namespace_checked(ok, Fun) -> Fun(#{}); own_namespace_checked({error, Reason}, _Fun) -> {error, {provider_authorization, Reason}}. resolve_org_directory(Io, Pool, Realm, Org, Procedure, NodeId, Fun) -> Find = maps:get(find_record, Io), case Find(Pool, macula_record:org_directory_key(Realm, Org)) of {ok, OrgDir} -> #{org_key := OrgKeyId} = macula_record:read_org_directory(OrgDir), resolve_delegation(Io, Pool, Realm, Procedure, OrgDir, OrgKeyId, NodeId, Fun); {error, not_found} -> {error, {provider_authorization, {org_directory, not_found}}}; {error, _} = E -> {error, {provider_authorization, E}} end. resolve_delegation(Io, Pool, Realm, Procedure, OrgDir, OrgKeyId, NodeId, Fun) -> Find = maps:get(find_record, Io), case Find(Pool, macula_record:procedure_delegation_key(OrgKeyId, NodeId)) of {ok, Deleg} -> sign_provider_advertisement(Io, Pool, Realm, Procedure, OrgDir, Deleg, NodeId, Fun); {error, not_found} -> {error, {provider_authorization, {procedure_delegation, not_found}}}; {error, _} = E -> {error, {provider_authorization, E}} end. %% The advertisement is signed under a bound: it must not outlive what %% authorizes it. An advertisement is built with its type's own %% lifetime, which knows nothing about the chain it carries, and both %% verifiers refuse one that ends after the earlier of the org %% directory's and the delegation's expiry — %% `macula_record:delegation_matched/3' here, and macula-station's at %% admission. Without the bound a provider spends the last stretch of %% every authorization window signing advertisements its own pool %% refuses, and goes dark before its delegation expires. sign_provider_advertisement(Io, Pool, Realm, Procedure, OrgDir, Deleg, NodeId, Fun) -> Authorization = #{org_directory => macula_record:encode(OrgDir), procedure_delegation => macula_record:encode(Deleg)}, Unsigned = macula_record:procedure_advertisement( NodeId, Realm, Procedure, NodeId, #{authorization => Authorization}), NotAfter = min(macula_record:expires_at(OrgDir), macula_record:expires_at(Deleg)), case (maps:get(sign_node_record, Io))(Pool, Unsigned, #{not_after => NotAfter}) of {ok, Signed} -> trusted_provider_advertisement( Io, Pool, Realm, Signed, fun(Verified) -> #{authorization := A} = macula_record:read_procedure_advertisement(Verified), Fun(#{authorization => A, not_after => NotAfter}) end); {error, _} = E -> {error, {provider_authorization, E}} end. %% The advertisement goes out only after its authorization verifies %% against the realm key this pool pins — a chain the pool cannot %% check never reaches a station. trusted_provider_advertisement(Io, Pool, Realm, Signed, Fun) -> case (maps:get(realm_key, Io))(Pool, Realm) of {ok, RealmKey} -> {ok, Profile} = macula_crypto_profile:configured(), Trust = #{profile => Profile, realm_key => RealmKey}, case macula_record:verify_authorization( Signed, Trust, erlang:system_time(millisecond)) of ok -> Fun(Signed); {error, _} = E -> {error, {provider_authorization, E}} end; none -> {error, {provider_authorization, no_realm_key}} end. %% @doc Stop advertising a LOCAL streaming procedure. -spec unadvertise_stream(procedure()) -> ok. unadvertise_stream(Procedure) when is_binary(Procedure) -> macula_stream_local:unadvertise(Procedure). %% @doc Stop advertising a streaming procedure on a V2 pool. -spec unadvertise_stream(pool(), realm(), procedure()) -> ok. unadvertise_stream(Pool, Realm, Procedure) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure) -> macula_client:unadvertise_stream(Pool, Realm, Procedure). %% @doc Send a binary chunk on the stream. -spec send(stream(), binary()) -> ok | {error, term()}. send(Stream, Bin) when is_pid(Stream), is_binary(Bin) -> macula_stream:send(Stream, Bin). %% @doc Send a chunk with explicit encoding. -spec send(stream(), binary() | term(), raw | msgpack) -> ok | {error, term()}. send(Stream, Body, Encoding) when is_pid(Stream) -> macula_stream:send(Stream, Body, Encoding). %% @doc Receive the next chunk (blocks). -spec recv(stream()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Stream) when is_pid(Stream) -> macula_stream:recv(Stream). -spec recv(stream(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Stream, Timeout) when is_pid(Stream) -> macula_stream:recv(Stream, Timeout). %% @doc The caller's seal report of a stream it opened (DESIGN_E2E_SEAL_REPORT): `sealed' 1 with the id of the KEM %% key the stream is sealed to, or 0 for a clear stream, and `provider', the stream's target. It states that the %% mechanism ran on this exchange, nothing more. It settles on the provider's first chunk or reply opened under the %% stream's key (on a clear stream, its first chunk, reply or end); after a reseal it names the reseal's key. Before %% it settles, on a stream that ended first, an error included, and on a local in-process stream it is %% `{error, not_settled}'; on a served stream, `{error, not_a_caller}'. It asks the stream process, so it answers %% while that process lives: until its owner ends, or the stream is closed. -spec stream_report(stream()) -> {ok, macula_station_link:report()} | {error, not_settled | not_a_caller}. stream_report(Stream) when is_pid(Stream) -> macula_stream:report(Stream). %% @doc Close a V1 stream (both sides). Renamed from `close/1' in %% 3.11.0 because `close/1' now refers to the V2 pool surface. -spec close_stream(stream()) -> ok. close_stream(Stream) when is_pid(Stream) -> macula_stream:close(Stream). %% @doc Half-close the write side; recv still drains. -spec close_send(stream()) -> ok. close_send(Stream) when is_pid(Stream) -> macula_stream:close_send(Stream). %% @doc Wait for the terminal reply (client-stream / bidi). -spec await_reply(stream()) -> {ok, term()} | {error, term()}. await_reply(Stream) when is_pid(Stream) -> macula_stream:await_reply(Stream). -spec await_reply(stream(), timeout()) -> {ok, term()} | {error, term()}. await_reply(Stream, Timeout) when is_pid(Stream) -> macula_stream:await_reply(Stream, Timeout). %% @doc Server-side: emit the terminal reply value. -spec set_reply(stream(), term()) -> ok. set_reply(Stream, Result) when is_pid(Stream) -> macula_stream:set_reply(Stream, Result). %% @doc Abort the stream with an error frame. -spec abort(stream(), binary(), binary()) -> ok | {error, {text_too_long | invalid_text, code}}. abort(Stream, Code, Message) when is_pid(Stream), is_binary(Code), is_binary(Message) -> macula_stream:abort(Stream, Code, Message). %%%=================================================================== %%% Cluster (LAN) %%%=================================================================== %% @doc Ensure this node is running in distributed mode. -spec ensure_distributed() -> ok | {error, term()}. ensure_distributed() -> macula_cluster:ensure_distributed(). %% @doc Subscribe to node up/down events. -spec monitor_nodes() -> ok. monitor_nodes() -> macula_cluster:monitor_nodes(). %% @doc Unsubscribe from node up/down events. -spec unmonitor_nodes() -> ok. unmonitor_nodes() -> macula_cluster:unmonitor_nodes(). %%%=================================================================== %%% Mesh Distribution %%%=================================================================== %% @doc Join the Macula relay mesh with Erlang distribution. %% %% After calling this, standard OTP distribution works across firewalls. %% `Opts' takes: %%
    %%
  • `relays' (required): the V2 pool's seeds, each a map with `host', %% `port' and `expected_node_id', the relay's 32-byte node_id, which %% every dial checks (D16). A relay without one refuses the join with %% `{error, {relays, expected_node_id_required}}' before any pool %% starts.
    • %%
  • `node_identity': the V2 pool's node identity key, %% `macula_node_keys:node_key()'. Default: the node's one stored %% identity, as for `connect/2'.
    • %%
%% %% Internally builds a V2 `macula_client:pool()' and registers it %% with `macula_dist_pool' as the carrier for `_dist.tunnel.*' %% traffic. Dist tunnel frames travel under the all-zeros realm %% (protocol-internal infrastructure, not bound to any user realm). -spec join_mesh(map()) -> ok | {error, term()}. join_mesh(Opts) -> joined(join_pool_args(Opts)). joined({ok, Relays, PoolOpts}) -> on_pool_for_join(macula_client:connect(Relays, PoolOpts)); joined({error, _} = Refusal) -> Refusal. %% The seeds and options of the pool a join starts: the relays, when every one names the node_id it expects, and the %% node identity key when one is given. join_pool_args(#{relays := Relays} = Opts) -> joinable(seeds_checked(Relays, Opts), Relays, Opts). %% A relay carrying `pin_tls_cert => true' or `verify' is refused here %% rather than at link start, so a join reports it instead of leaving a pool respawning a %% seed that can never start. joinable(ok, Relays, Opts) -> pinned_relays(lists:all(fun pinned_relay/1, Relays), Relays, maps:with([node_identity], Opts)); joinable({error, _} = Refusal, _Relays, _Opts) -> Refusal. pinned_relays(true, Relays, PoolOpts) -> {ok, Relays, PoolOpts}; pinned_relays(false, _Relays, _PoolOpts) -> {error, {relays, expected_node_id_required}}. pinned_relay(#{host := _, port := _, expected_node_id := <<_:256>>}) -> true; pinned_relay(_Relay) -> false. on_pool_for_join({ok, Pool}) -> wait_for_pool(Pool, 30), os:putenv("MACULA_DIST_MODE", "relay"), macula_dist_pool:register_mesh_pool(Pool), macula_dist_pool:advertise_dist_accept(), ?LOG_INFO("[macula] Joined mesh — distribution enabled"), ok; on_pool_for_join({error, Reason}) -> ?LOG_ERROR("[macula] Failed to join mesh: ~p", [Reason]), {error, Reason}. %% @doc Enable Erlang distribution over a dedicated dist relay %% (`macula-io/macula-dist-relay'). %% %% Different from `join_mesh/1': %% - Connects to a dist relay (port 4434, ALPN `macula-dist'), NOT the %% pub/sub station mesh %% - No mesh_client, no pub/sub subscriptions — only dist traffic %% - Uses raw QUIC stream routing with no MessagePack overhead %% %% Options: %% - `url' (required): `<<"quic://relay.example.com:4434">>' %% %% After this returns `ok', standard OTP distribution (`rpc:call/4', %% `gen_server:call/3' across nodes, `pg' groups, etc.) works across %% firewalls via the dist relay. %% %% The relay client runs as a temporary child of the macula application %% supervisor. It does not reconnect: when the relay closes the %% connection the client ends and is not restarted. Monitor the pid from %% `dist_relay_client/0' to learn that, then call this function again. %% Returns `{error, macula_not_started}' when the macula application is %% not running. -spec join_dist_relay(map()) -> ok | {error, term()}. join_dist_relay(Opts) -> Url = maps:get(url, Opts), NodeName = atom_to_binary(node()), start_dist_relay_client(whereis(macula_root), Url, NodeName). start_dist_relay_client(undefined, _Url, _NodeName) -> {error, macula_not_started}; start_dist_relay_client(_Root, Url, NodeName) -> ChildSpec = macula_dist_relay_client:child_spec(Url, NodeName), case supervisor:start_child(macula_root, ChildSpec) of {ok, _Pid} -> os:putenv("MACULA_DIST_MODE", "dist_relay"), ?LOG_INFO("[macula] Joined dist relay ~s, distribution enabled", [Url]), ok; {error, {already_started, _Pid}} -> os:putenv("MACULA_DIST_MODE", "dist_relay"), ?LOG_INFO("[macula] dist_relay_client already running, mode set"), ok; {error, Reason} -> Refusal = child_start_refusal(Reason), ?LOG_ERROR("[macula] Failed to join dist relay: ~p", [Refusal]), {error, Refusal} end. %% `supervisor:start_child/2' wraps a child's own refusal with the child %% spec that failed to start. A caller of `join_dist_relay/1' wants the %% refusal, not the spec it already gave us: the client refuses when its %% operator has not accepted an unidentified peer %% (`macula_dist:unidentified_peer_refusal/0'), and that term says so. child_start_refusal({Refusal, Child}) when is_tuple(Child), element(1, Child) =:= child -> Refusal; child_start_refusal(Reason) -> Reason. %% @doc The dist relay client that `join_dist_relay/1' started, if it %% is running. %% %% The client exits with `{relay_closed, Reason}' when the relay closes %% the connection and is not restarted. Monitor the returned pid and call %% `join_dist_relay/1' again after it goes down. -spec dist_relay_client() -> {ok, pid()} | {error, not_joined}. dist_relay_client() -> dist_relay_client_result(macula_dist_relay_client:whereis_client()). dist_relay_client_result(undefined) -> {error, not_joined}; dist_relay_client_result(Pid) -> {ok, Pid}. %% @private Wait until the V2 pool has at least one healthy %% station_link (CONNECT/HELLO completed). One-second polling, capped %% at `Retries' iterations. A handshake, not a route: see `healthy_links' %% in `macula_client:status/1'. wait_for_pool(_Pool, 0) -> ?LOG_WARNING("[macula] Mesh pool not ready after timeout"); wait_for_pool(Pool, Retries) -> on_pool_status(macula_client:status(Pool), Pool, Retries). on_pool_status({ok, #{healthy_links := N}}, _Pool, _Retries) when N > 0 -> ?LOG_INFO("[macula] Mesh pool connected (~p healthy link(s))", [N]), ok; on_pool_status(_Other, Pool, Retries) -> timer:sleep(1000), wait_for_pool(Pool, Retries - 1). %%%=================================================================== %%% Peer-supplied maps (D26) %%%=================================================================== %% @doc A field of a map a peer supplied, or `undefined' when it is absent. See `field/3'. -spec field(atom() | binary(), map()) -> term(). field(Name, Map) -> field(Name, Map, undefined). %% @doc A field of a map a peer supplied (D26), or `Default' when it is absent. A map from the codec carries its text %% keys as `{text, Bin}'; a map handed over in process may carry atom or binary keys. The lookup tries `{text, Name}', %% then the atom, then the binary, so a handler reads both kinds of map the same way. Looking up a binary name never %% creates an atom. -spec field(atom() | binary(), map(), term()) -> term(). field(Name, Map, Default) when (is_atom(Name) orelse is_binary(Name)), is_map(Map) -> first_found(field_keys(Name), Map, Default). %% @doc The binary of a text value a peer supplied (D26): the binary of `{text, Bin}', a binary unchanged, and %% `badarg' for anything else. -spec text({text, binary()} | binary()) -> binary(). text({text, Bin}) when is_binary(Bin) -> Bin; text(Bin) when is_binary(Bin) -> Bin; text(Other) -> erlang:error(badarg, [Other]). field_keys(Name) when is_atom(Name) -> Bin = atom_to_binary(Name), [{text, Bin}, Name, Bin]; field_keys(Name) when is_binary(Name) -> [{text, Name}] ++ existing_atom(Name) ++ [Name]. existing_atom(Name) -> try [binary_to_existing_atom(Name)] catch error:badarg -> [] end. first_found([], _Map, Default) -> Default; first_found([Key | Keys], Map, Default) -> found(maps:find(Key, Map), Keys, Map, Default). found({ok, Value}, _Keys, _Map, _Default) -> Value; found(error, Keys, Map, Default) -> first_found(Keys, Map, Default). %%------------------------------------------------------------------ %% `pin_tls_cert' and `verify' are refused, not accepted and ignored %%------------------------------------------------------------------ %% Every public entry point that takes a seed, a station or a per-dial TLS %% trust map runs this before it does anything else: `connect/2', %% `call_station/7,8', `call_stream_station/7' and `join_mesh/1'. %% `pin_tls_cert => true' is refused, and `verify' in any value; %% `pin_tls_cert => false' and absence %% pass through. %% %% WHY NEITHER CAN BE HONOURED, so the next reader does not re-litigate it. %% A dial trusts a station in exactly one way (plan decisions D12 and D16): %% the station presents one self-signed ML-DSA-87 certificate on its TLS %% key, its handshake signature must verify under that key, and its %% identity is proved by the signed handshake against `expected_node_id', %% which is mandatory on a client dial. `pin_tls_cert' once mapped %% `expected_node_id' onto a pinned Ed25519 key, which stopped meaning %% anything when a node_id became a hash over an ML-DSA-87 identity key and %% the TLS key became a key of its own; the pin primitive is gone from the %% QUIC NIF. `verify' chose between a webpki chain check, which no ML-DSA %% certificate can pass because no authority issues one, and no check at %% all, which the NIF no longer offers. %% %% So the choice is between accepting a value that does nothing and saying %% so. An inert security option is worse than an absent one, because a %% reader takes it for a check that is happening: that is macula#15, and it %% reached hexdocs as a documented default of `true' that was never true. %% `pin_tls_cert => false' still passes, as it did through 11.x; `verify' %% is refused in any value, since 12.0.0 is where it went. %% %% ⚠ EVERY map a caller can put the keys in, not just `Opts'. The seed and %% station maps are where the options historically lived, so checking only %% `Opts' advertises a check that does not happen on the path a reader of %% our own CHANGELOG would take. %% %% The refusal is wrapped in `refused' because nothing is sent and every %% candidate refuses it identically, which is `request' scope in the dial %% taxonomy. Left bare it would fall to a catch-all and be classified %% `provider', meaning "may have reached a provider", which is the wrong %% scope for a refusal that sent nothing. %% %% ⚠ MAPS ONLY. A seed may be a charlist (`seed()' includes `string()'), %% so an `is_list' clause here would walk `"quic://..."' character by %% character and be correct only by accident, every integer falling to the %% catch-all. The list is mapped over by `seeds_checked/2', which knows it %% has a list OF seeds rather than a seed that happens to be a list. -spec trust_options_checked(map() | macula_client:seed()) -> ok | {error, {refused, {pin_tls_cert | verify, atom()}}}. trust_options_checked(#{pin_tls_cert := true}) -> {error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}}; trust_options_checked(#{verify := _}) -> {error, {refused, {verify, one_verification_mode}}}; trust_options_checked(_NoRefusedOption) -> ok. %% `connect/2' and `join_mesh/1': a LIST of seeds, plus the option map. -spec seeds_checked([macula_client:seed()], map()) -> ok | {error, {refused, {pin_tls_cert | verify, atom()}}}. seeds_checked(Seeds, Opts) when is_list(Seeds) -> first_error([trust_options_checked(S) || S <- Seeds] ++ [trust_options_checked(Opts)]). %% One station or seed, plus the option map, in the order the public %% functions receive them. -spec target_checked(macula_client:seed(), map()) -> ok | {error, {refused, {pin_tls_cert | verify, atom()}}}. target_checked(Target, Opts) -> first_error([trust_options_checked(Target), trust_options_checked(Opts)]). %% Named `first_error' and not `first_refusal': `macula_station_link' has a %% `first_refusal/1' of its own with a different shape (it folds `none' and %% a refusal term, not `ok' and an `{error, _}'). Two functions of the same %% name meaning different things in one module family is the small version %% of a contract in two places. first_error([{error, _} = Refusal | _Rest]) -> Refusal; first_error([ok | Rest]) -> first_error(Rest); first_error([]) -> ok. %% Run the call, or return the refusal that stopped it. refused(ok, Call) -> Call(); refused({error, _} = Refusal, _Call) -> Refusal.