Packages
macula
4.2.1
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/host_attach_controller/macula_host_attach_controller.erl
%%%-------------------------------------------------------------------
%%% @doc Host-side controller for the macula-net attach plane.
%%%
%%% PLAN_MACULA_NET_PHASE3_5_TRANSPORT_SEAM.md §5. When a station
%%% hosts daemons, this slice is the transport's inbound handler. It
%%% owns three responsibilities:
%%%
%%% <ol>
%%% <li><b>attach_v1 dispatch</b> — On a `macula_attach_v1' frame,
%%% parse + validate the daemon's signed delegation, then call
%%% the configured `attach_fn' with the originating
%%% <em>StreamRef</em> as the `attach_conn'. Owning the StreamRef
%%% at attach time is what lets the host send replies on the
%%% same bidi stream the daemon dialed in on.</li>
%%% <li><b>hosted-data forwarding</b> — On a `data' frame whose `dst'
%%% is in the host_identity table, look up the stored StreamRef
%%% and re-send the original CBOR on it via `attach_send_fn'.
%%% The dst is checked by ETS membership lookup, not by trusting
%%% the daemon's claim — the address-binding-by-construction
%%% discipline from the resolver applies here too.</li>
%%% <li><b>fallback</b> — Anything else (non-data envelopes, data
%%% for non-hosted dst, malformed CBOR) is handed to `fallback_fn',
%%% typically {@link macula_deliver_packet:handle_envelope/1}.</li>
%%% </ol>
%%%
%%% Every external coupling is a callback in the config; production
%%% wiring lives in {@link macula_net} (host-mode boot path), tests
%%% pass capture functions to a process mailbox.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_host_attach_controller).
-behaviour(gen_server).
-export([
start_link/1,
stop/0,
handle/2,
delegation_from_wire/1
]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-export_type([config/0]).
-define(SERVER, ?MODULE).
-define(FRAME_HEADER_BYTES, 4).
-type stream_ref() :: macula_net_transport:stream_ref().
-type attach_send_fn() :: fun((stream_ref(), iodata()) -> ok | {error, term()}).
-type lookup_fn() :: fun((<<_:128>>) -> {ok, stream_ref()} | not_found).
-type attach_fn() :: fun((<<_:128>>, <<_:256>>,
macula_record:host_delegation(),
stream_ref()) -> ok | {error, term()}).
-type fallback_fn() :: fun((binary()) -> any()).
-type forward_fn() :: fun((Cbor :: binary(), Dst :: <<_:128>>) ->
{ok, term()} | {error, term()}).
-type config() :: #{
realm_pubkey := <<_:256>>,
host_pubkey := <<_:256>>,
attach_send_fn := attach_send_fn(),
lookup_fn := lookup_fn(),
attach_fn := attach_fn(),
fallback_fn := fallback_fn(),
forward_fn => forward_fn()
}.
-record(state, {
config :: config(),
own_addr :: <<_:128>>
}).
%% =============================================================================
%% Public API
%% =============================================================================
-spec start_link(config()) -> {ok, pid()} | {error, term()}.
start_link(Config) ->
gen_server:start_link({local, ?SERVER}, ?MODULE, Config, []).
-spec stop() -> ok.
stop() ->
case whereis(?SERVER) of
undefined -> ok;
_ -> gen_server:stop(?SERVER)
end.
%% @doc Transport handler entry point. Suitable for direct use as the
%% argument to {@link macula_net_transport_quic:set_handler/1} when the
%% station is configured in host mode.
-spec handle(binary(), stream_ref()) -> ok.
handle(Cbor, StreamRef) when is_binary(Cbor) ->
gen_server:cast(?SERVER, {frame, Cbor, StreamRef}).
%% @doc Convert the wire-shape (single-letter keys) delegation map into
%% the in-process atom-keyed form that
%% {@link macula_record:verify_host_delegation/1} expects. Exported for
%% test reuse.
-spec delegation_from_wire(map()) -> macula_record:host_delegation() | error.
delegation_from_wire(#{<<"d">> := D, <<"h">> := H, <<"r">> := R,
<<"nb">> := NB, <<"na">> := NA, <<"s">> := S}) ->
#{daemon_pubkey => D, host_pubkey => H, realm_pubkey => R,
not_before_ms => NB, not_after_ms => NA, daemon_sig => S};
delegation_from_wire(_) ->
error.
%% =============================================================================
%% gen_server callbacks
%% =============================================================================
init(#{realm_pubkey := Realm, host_pubkey := HostPk,
attach_send_fn := S, lookup_fn := L,
attach_fn := A, fallback_fn := F} = Config)
when is_function(S, 2), is_function(L, 1),
is_function(A, 4), is_function(F, 1) ->
process_flag(trap_exit, true),
OwnAddr = macula_address:derive(Realm, HostPk),
%% Default forward_fn: drop. Production wires
%% fun macula_route_packet:dispatch_envelope/2 here. Tests pass a
%% capture fn. With no forward_fn the controller falls through to
%% fallback_fn for non-hosted-non-local data, preserving the
%% Phase 3.5 behaviour for callers that don't opt in.
{ok, #state{config = Config, own_addr = OwnAddr}}.
handle_call(_Other, _From, State) ->
{reply, {error, unknown_call}, State}.
handle_cast({frame, Cbor, StreamRef}, State) ->
dispatch_decoded(macula_cbor_nif:unpack(Cbor), Cbor, StreamRef, State),
{noreply, State};
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info(_Msg, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%% =============================================================================
%% Dispatch — multi-clause, no nesting
%% =============================================================================
dispatch_decoded({ok, Map}, Cbor, StreamRef, State) when is_map(Map) ->
dispatch_typed(Map, Cbor, StreamRef, State);
dispatch_decoded(_Other, Cbor, _StreamRef, State) ->
fallback(Cbor, State).
dispatch_typed(#{<<"type">> := <<"macula_attach_v1">>} = M, _Cbor, StreamRef, State) ->
handle_attach(M, StreamRef, State);
dispatch_typed(#{<<"type">> := <<"data">>, <<"dst">> := Dst} = _M, Cbor, _StreamRef, State) ->
route_data(safe_lookup(Dst, State), Dst, Cbor, State);
dispatch_typed(_Other, Cbor, _StreamRef, State) ->
fallback(Cbor, State).
%% --- attach -----------------------------------------------------------------
handle_attach(#{<<"daemon_pubkey">> := DaemonPk,
<<"daemon_addr">> := DaemonAddr,
<<"delegation">> := Wire} = _Map,
StreamRef, State)
when is_binary(DaemonPk), byte_size(DaemonPk) =:= 32,
is_binary(DaemonAddr), byte_size(DaemonAddr) =:= 16 ->
attach_with_delegation(delegation_from_wire(Wire),
DaemonAddr, DaemonPk, StreamRef, State);
handle_attach(_Other, _StreamRef, _State) ->
ok.
attach_with_delegation(error, _Addr, _Pk, _StreamRef, _State) ->
ok;
attach_with_delegation(Delegation, Addr, Pk, StreamRef,
#state{config = #{attach_fn := AttachFn}}) ->
log_attach(safe_call4(AttachFn, Addr, Pk, Delegation, StreamRef), Addr).
log_attach(ok, Addr) ->
telemetry:execute([macula, net, attach, attached],
#{count => 1},
#{daemon_addr_prefix => addr_prefix_hex(Addr)}),
ok;
log_attach({error, Reason}, Addr) ->
telemetry:execute([macula, net, attach, detached],
#{count => 1},
#{event => <<"attach_rejected">>,
reason => reason_bin(Reason),
daemon_addr_prefix => addr_prefix_hex(Addr)}),
error_logger:warning_msg(
"[host_attach_controller] attach rejected for ~p: ~p",
[binary:part(Addr, 0, 8), Reason]),
ok.
addr_prefix_hex(Addr) when is_binary(Addr), byte_size(Addr) >= 8 ->
<<Prefix:8/binary, _/binary>> = Addr,
bin_to_hex(Prefix);
addr_prefix_hex(_) -> <<"unknown">>.
bin_to_hex(Bin) ->
list_to_binary([io_lib:format("~2.16.0b", [B]) || <<B>> <= Bin]).
reason_bin(R) when is_atom(R) -> atom_to_binary(R, utf8);
reason_bin(R) when is_binary(R) -> R;
reason_bin(_) -> <<"unknown">>.
%% --- data routing -----------------------------------------------------------
%%
%% Three-way split per the §7.1 sub-spec:
%% 1. dst is a hosted daemon address — forward on its attach stream.
%% 2. dst is the host station's own address — fallback (deliver_packet
%% writes to TUN).
%% 3. otherwise — forward via route_packet:dispatch_envelope.
route_data({ok, OutStream}, _Dst, Cbor, #state{config = #{attach_send_fn := SendFn}}) ->
%% Re-frame: the daemon expects the same length-prefixed CBOR shape
%% the station wire uses.
Frame = <<(byte_size(Cbor)):32/big, Cbor/binary>>,
log_send(safe_call2(SendFn, OutStream, Frame));
route_data(not_found, Dst, Cbor, #state{own_addr = OwnAddr} = State)
when Dst =:= OwnAddr ->
fallback(Cbor, State);
route_data(not_found, Dst, Cbor, State) ->
forward_or_fallback(Dst, Cbor, State).
forward_or_fallback(Dst, Cbor, #state{config = Config} = State) ->
forward_dispatch(maps:get(forward_fn, Config, undefined), Dst, Cbor, State).
forward_dispatch(undefined, _Dst, Cbor, State) ->
fallback(Cbor, State);
forward_dispatch(Fn, Dst, Cbor, _State) ->
log_forward(safe_call2(Fn, Cbor, Dst)).
log_send(ok) -> ok;
log_send({error, Reason}) ->
error_logger:warning_msg(
"[host_attach_controller] send to hosted daemon failed: ~p",
[Reason]),
ok.
log_forward({ok, _}) -> ok;
log_forward(ok) -> ok;
log_forward({error, Reason}) ->
error_logger:warning_msg(
"[host_attach_controller] forward failed: ~p", [Reason]),
ok.
%% --- fallback ---------------------------------------------------------------
fallback(Cbor, #state{config = #{fallback_fn := Fun}}) ->
safe_call1(Fun, Cbor),
ok.
%% =============================================================================
%% Boundary helpers — user callbacks must not crash the controller.
%% =============================================================================
safe_lookup(Dst, #state{config = #{lookup_fn := L}}) ->
try L(Dst)
catch _:_ -> not_found
end.
safe_call1(F, A) ->
try F(A)
catch _:_ -> ok
end.
safe_call2(F, A, B) ->
try F(A, B)
catch _:_ -> {error, callback_crashed}
end.
safe_call4(F, A, B, C, D) ->
try F(A, B, C, D)
catch _:_ -> {error, callback_crashed}
end.