Current section
Files
Jump to
Current section
Files
src/protocols/http/erlang_adk_http.erl
%%%-------------------------------------------------------------------
%% @doc Supervised lifecycle owner for the optional project HTTP endpoint.
%% @end
%%%-------------------------------------------------------------------
-module(erlang_adk_http).
-behaviour(gen_server).
-export([start_link/0]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
-define(LISTENER, erlang_adk_a2a_http).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
init([]) ->
process_flag(trap_exit, true),
case listener_config() of
{ok, Config} -> start_listener(Config);
{error, Reason} -> {stop, Reason}
end.
handle_call(_Request, _From, State) ->
{reply, {error, unsupported_call}, State}.
handle_cast(_Request, State) ->
{noreply, State}.
handle_info({'DOWN', MonitorRef, process, ListenerPid, Reason},
#{listener_pid := ListenerPid,
monitor_ref := MonitorRef} = State) ->
{stop, {http_listener_terminated, Reason}, State};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, #{monitor_ref := MonitorRef}) ->
erlang:demonitor(MonitorRef, [flush]),
_ = cowboy:stop_listener(?LISTENER),
ok;
terminate(_Reason, _State) ->
ok.
code_change(_OldVersion, State, _Extra) ->
{ok, State}.
%% internal functions
start_listener(Config) ->
MaxBodyBytes = maps:get(max_body_bytes, Config),
Routes = legacy_a2a_routes(Config, MaxBodyBytes) ++
a2a_v1_routes(Config) ++ dev_routes(Config),
Dispatch = cowboy_router:compile([{'_', Routes}]),
SocketOptions = [{ip, maps:get(ip, Config)},
{port, maps:get(port, Config)}] ++
tls_socket_options(Config),
TransportOptions =
#{socket_opts => SocketOptions,
num_acceptors => maps:get(num_acceptors, Config),
max_connections => maps:get(max_connections, Config)},
ProtocolOptions =
#{env => #{dispatch => Dispatch},
idle_timeout => maps:get(idle_timeout, Config),
request_timeout => maps:get(request_timeout, Config),
max_keepalive => maps:get(max_keepalive, Config)},
case start_cowboy(Config, TransportOptions, ProtocolOptions) of
{ok, ListenerPid} ->
MonitorRef = erlang:monitor(process, ListenerPid),
{ok, #{listener_pid => ListenerPid,
monitor_ref => MonitorRef}};
{error, Reason} ->
{stop, {http_listener_start_failed, Reason}}
end.
listener_config() ->
Specs = [
{port, a2a_port, 8080, fun valid_port/1},
{ip, a2a_ip, {127, 0, 0, 1}, fun valid_ip/1},
{num_acceptors, a2a_num_acceptors, 10, fun positive_integer/1},
{max_connections, a2a_max_connections, 1024,
fun positive_integer/1},
{max_body_bytes, a2a_max_body_bytes, 1048576,
fun positive_integer/1},
{request_timeout, a2a_request_timeout, 10000,
fun positive_integer/1},
{idle_timeout, a2a_idle_timeout, 60000,
fun positive_integer/1},
{max_keepalive, a2a_max_keepalive, 100,
fun non_negative_integer/1},
{allow_non_loopback, a2a_allow_non_loopback, false,
fun erlang:is_boolean/1},
{trusted_tls_proxy, a2a_trusted_tls_proxy, false,
fun erlang:is_boolean/1},
{tls_options, a2a_tls_options, undefined,
fun valid_tls_options/1}
],
case read_config(Specs, #{}) of
{ok, Config} -> add_endpoint_config(Config);
{error, _} = Error -> Error
end.
add_endpoint_config(Config) ->
A2AEnabled = application:get_env(erlang_adk, a2a_enabled, false),
A2AV1Enabled = application:get_env(erlang_adk, a2a_v1_enabled, false),
DevEnabled = application:get_env(erlang_adk, dev_enabled, false),
case is_boolean(A2AEnabled) andalso is_boolean(A2AV1Enabled)
andalso is_boolean(DevEnabled) of
false -> {error, invalid_http_endpoint_config};
true ->
Config1 = Config#{a2a_enabled => A2AEnabled,
a2a_v1_enabled => A2AV1Enabled,
dev_enabled => DevEnabled},
case add_a2a_v1_config(Config1) of
{ok, Config2} ->
case validate_public_route_mix(Config2) of
ok ->
case validate_dev_exposure(Config2) of
ok ->
case validate_legacy_exposure(Config2) of
ok ->
ConfigResult = case DevEnabled of
false -> {ok, Config2};
true -> add_dev_config(Config2)
end,
case ConfigResult of
{ok, Config3} ->
validate_exposure(Config3);
{error, _} = Error -> Error
end;
{error, _} = Error -> Error
end;
{error, _} = Error -> Error
end;
{error, _} = Error -> Error
end;
{error, _} = Error -> Error
end
end.
add_a2a_v1_config(#{a2a_v1_enabled := false} = Config) ->
{ok, Config};
add_a2a_v1_config(Config) ->
Auth = application:get_env(erlang_adk, a2a_v1_auth, none),
MaxBody = application:get_env(
erlang_adk, a2a_v1_max_body_bytes,
maps:get(max_body_bytes, Config)),
Heartbeat = application:get_env(
erlang_adk, a2a_v1_sse_heartbeat_ms, 15000),
MaxExtensions = application:get_env(
erlang_adk, a2a_v1_max_extensions, 32),
MaxExtensionHeaderBytes = application:get_env(
erlang_adk,
a2a_v1_max_extension_header_bytes, 8192),
AuthTimeout = application:get_env(
erlang_adk, a2a_v1_auth_timeout_ms, 5000),
AuthMaxHeap = application:get_env(
erlang_adk, a2a_v1_auth_max_heap_words, 300000),
case valid_a2a_v1_auth(Auth) andalso positive_integer(MaxBody)
andalso positive_integer(Heartbeat)
andalso positive_integer(MaxExtensions)
andalso MaxExtensions =< 128
andalso positive_integer(MaxExtensionHeaderBytes)
andalso MaxExtensionHeaderBytes =< 65536
andalso positive_integer(AuthTimeout)
andalso AuthTimeout =< 30000
andalso is_integer(AuthMaxHeap)
andalso AuthMaxHeap >= 1000
andalso AuthMaxHeap =< 2000000 of
true ->
{ok, Config#{a2a_v1_config =>
#{server => adk_a2a_v1_server,
auth => Auth,
max_body_bytes => MaxBody,
sse_heartbeat_ms => Heartbeat,
max_extensions => MaxExtensions,
max_extension_header_bytes =>
MaxExtensionHeaderBytes,
auth_timeout_ms => AuthTimeout,
auth_max_heap_words => AuthMaxHeap}}};
false -> {error, invalid_a2a_v1_endpoint_config}
end.
add_dev_config(Config) ->
case developer_token() of
{ok, Token} ->
DevConfig = #{auth_token => Token,
session_service => application:get_env(
erlang_adk,
dev_session_service,
erlang_adk_session),
runner_options => application:get_env(
erlang_adk,
dev_runner_options, #{}),
run_options => application:get_env(
erlang_adk, dev_run_options, #{}),
resource_provider => application:get_env(
erlang_adk,
dev_resource_provider,
undefined),
max_resource_results => application:get_env(
erlang_adk,
dev_max_resource_results,
100),
diagnostic_timeout_ms => application:get_env(
erlang_adk,
dev_diagnostic_timeout_ms,
5000),
diagnostic_context_policy => application:get_env(
erlang_adk,
dev_diagnostic_context_policy,
#{max_bytes => 1048576,
max_tokens => 262144,
overflow => truncate}),
max_body_bytes => application:get_env(
erlang_adk,
dev_max_body_bytes, 65536),
max_field_bytes => application:get_env(
erlang_adk,
dev_max_field_bytes, 4096),
sse_heartbeat_ms => application:get_env(
erlang_adk,
dev_sse_heartbeat_ms, 15000),
sse_max_events => application:get_env(
erlang_adk,
dev_sse_max_events, 128),
sse_max_bytes => application:get_env(
erlang_adk,
dev_sse_max_bytes, 1048576),
sse_max_duration_ms => application:get_env(
erlang_adk,
dev_sse_max_duration_ms,
300000),
max_session_results => application:get_env(
erlang_adk,
dev_max_session_results,
100),
live_principal => application:get_env(
erlang_adk,
dev_live_principal,
undefined),
live_credit => application:get_env(
erlang_adk,
dev_live_credit,
#{messages => 16,
bytes => 4194304})},
case adk_dev_router:validate_config(DevConfig) of
{ok, SafeDevConfig} ->
{ok, Config#{dev_config => SafeDevConfig}};
{error, Reason} ->
{error, {invalid_dev_platform_config, Reason}}
end;
{error, _} = Error -> Error
end.
developer_token() ->
case application:get_env(erlang_adk, dev_auth_token) of
{ok, Token} when is_binary(Token), byte_size(Token) >= 16 ->
{ok, Token};
{ok, _Invalid} ->
{error, invalid_dev_auth_token};
undefined ->
EnvName = application:get_env(
erlang_adk, dev_auth_token_env,
"ERLANG_ADK_DEV_TOKEN"),
case is_list(EnvName) andalso os:getenv(EnvName) of
TokenString when is_list(TokenString),
length(TokenString) >= 16 ->
{ok, unicode:characters_to_binary(TokenString)};
_ ->
{error, missing_dev_auth_token}
end
end.
legacy_a2a_routes(#{a2a_enabled := true}, MaxBodyBytes) ->
[{"/a2a/prompt", erlang_adk_a2a_handler,
#{max_body_bytes => MaxBodyBytes}}];
legacy_a2a_routes(_Config, _MaxBodyBytes) ->
[].
a2a_v1_routes(#{a2a_v1_enabled := true, a2a_v1_config := Config}) ->
[{"/.well-known/agent-card.json", adk_a2a_v1_handler,
Config#{endpoint => card}},
{"/a2a/v1", adk_a2a_v1_handler,
Config#{endpoint => jsonrpc}}];
a2a_v1_routes(_Config) -> [].
dev_routes(#{dev_enabled := true, dev_config := Config}) ->
%% validate_config/1 has already replaced the raw token with a digest.
adk_dev_router:routes_validated(Config);
dev_routes(_Config) ->
[].
read_config([], Config) ->
{ok, Config};
read_config([{Name, EnvKey, Default, Validator} | Rest], Config) ->
Value = application:get_env(erlang_adk, EnvKey, Default),
case Validator(Value) of
true -> read_config(Rest, Config#{Name => Value});
false -> {error, {invalid_application_env, EnvKey, Value}}
end.
valid_port(Value) ->
is_integer(Value) andalso Value > 0 andalso Value =< 65535.
valid_ip({A, B, C, D}) ->
lists:all(fun valid_ipv4_octet/1, [A, B, C, D]);
valid_ip({A, B, C, D, E, F, G, H}) ->
lists:all(fun valid_ipv6_segment/1, [A, B, C, D, E, F, G, H]);
valid_ip(_) ->
false.
valid_ipv4_octet(Value) ->
is_integer(Value) andalso Value >= 0 andalso Value =< 255.
valid_ipv6_segment(Value) ->
is_integer(Value) andalso Value >= 0 andalso Value =< 16#ffff.
positive_integer(Value) ->
is_integer(Value) andalso Value > 0.
non_negative_integer(Value) ->
is_integer(Value) andalso Value >= 0.
valid_a2a_v1_auth(none) -> true;
valid_a2a_v1_auth(Module) when is_atom(Module) -> true;
valid_a2a_v1_auth(Fun) when is_function(Fun, 3) -> true;
valid_a2a_v1_auth(_) -> false.
start_cowboy(#{tls_options := undefined}, Transport, Protocol) ->
cowboy:start_clear(?LISTENER, Transport, Protocol);
start_cowboy(_Config, Transport, Protocol) ->
cowboy:start_tls(?LISTENER, Transport, Protocol).
tls_socket_options(#{tls_options := undefined}) -> [];
tls_socket_options(#{tls_options := Options}) -> Options.
valid_tls_options(undefined) -> true;
valid_tls_options(Options) when is_list(Options), Options =/= [] ->
lists:all(fun(Option) -> is_tuple(Option) andalso tuple_size(Option) =:= 2
end, Options)
andalso not lists:keymember(ip, 1, Options)
andalso not lists:keymember(port, 1, Options)
andalso has_tls_identity(Options);
valid_tls_options(_) -> false.
has_tls_identity(Options) ->
(lists:keymember(certfile, 1, Options)
andalso lists:keymember(keyfile, 1, Options))
orelse (lists:keymember(cert, 1, Options)
andalso lists:keymember(key, 1, Options)).
validate_public_route_mix(#{a2a_v1_enabled := true,
ip := Ip,
a2a_enabled := Legacy,
dev_enabled := Dev}) ->
case is_loopback(Ip) orelse (Legacy =:= false andalso Dev =:= false) of
true -> ok;
false -> {error, public_a2a_v1_requires_dedicated_listener}
end;
validate_public_route_mix(_Config) -> ok.
%% The bundled developer console uses one shared administrator bearer and is
%% deliberately single-operator tooling. Enforce the loopback boundary in
%% the listener owner, not only in the CLI parser, so application startup and
%% direct erlang_adk_http:start_link/0 calls cannot expose /dev or co-host it
%% on a public legacy A2A socket.
validate_dev_exposure(#{dev_enabled := true, ip := Ip}) ->
case is_loopback(Ip) of
true -> ok;
false -> {error, developer_server_must_bind_loopback}
end;
validate_dev_exposure(_Config) -> ok.
%% The project-specific legacy prompt route has no authentication contract.
%% Keep it useful for local compatibility without allowing a configuration
%% change to turn it into an unauthenticated, billable public endpoint. The
%% public A2A v1 listener has its own authenticated TLS policy below.
validate_legacy_exposure(#{a2a_enabled := true, ip := Ip}) ->
case is_loopback(Ip) of
true -> ok;
false -> {error, legacy_a2a_server_must_bind_loopback}
end;
validate_legacy_exposure(_Config) -> ok.
validate_exposure(#{a2a_v1_enabled := false} = Config) -> {ok, Config};
validate_exposure(Config) ->
Auth = maps:get(auth, maps:get(a2a_v1_config, Config)),
Card0 = application:get_env(erlang_adk, a2a_v1_card, undefined),
case adk_a2a_v1_card:validate(Card0) of
{ok, Card} -> validate_card_and_transport(Config, Auth, Card);
{error, Reason} -> {error, {invalid_a2a_v1_card, Reason}}
end.
validate_card_and_transport(Config, none, Card) ->
case has_security_requirements(Card) of
true -> {error, a2a_v1_auth_card_mismatch};
false -> validate_anonymous_transport(Config)
end;
validate_card_and_transport(Config, _Auth, Card) ->
case is_loopback(maps:get(ip, Config)) of
true -> {ok, Config};
false ->
case has_security_declarations(Card) of
false -> {error, a2a_v1_auth_card_mismatch};
true -> validate_public_transport(Config, Card)
end
end.
validate_anonymous_transport(Config) ->
case is_loopback(maps:get(ip, Config)) of
true -> {ok, Config};
false -> {error, non_loopback_a2a_v1_requires_authentication}
end.
validate_public_transport(Config, Card) ->
Allow = maps:get(allow_non_loopback, Config),
HasTlsBoundary = maps:get(tls_options, Config) =/= undefined
orelse maps:get(trusted_tls_proxy, Config),
case {Allow, HasTlsBoundary, card_uses_https(Card)} of
{false, _, _} ->
{error, non_loopback_a2a_v1_requires_explicit_opt_in};
{true, false, _} ->
{error, non_loopback_a2a_v1_requires_tls};
{true, true, false} ->
{error, non_loopback_a2a_v1_requires_https_card};
{true, true, true} -> {ok, Config}
end.
has_security_requirements(Card) ->
case maps:get(<<"securityRequirements">>, Card, []) of
[_ | _] -> true;
_ -> false
end.
has_security_declarations(Card) ->
Schemes = maps:get(<<"securitySchemes">>, Card, #{}),
is_map(Schemes) andalso map_size(Schemes) > 0
andalso has_security_requirements(Card).
card_uses_https(Card) ->
case adk_a2a_v1_card:jsonrpc_interface(Card) of
{ok, Interface} ->
try uri_string:parse(maps:get(<<"url">>, Interface)) of
Parsed ->
to_binary(maps:get(scheme, Parsed, <<>>)) =:= <<"https">>
catch _:_ -> false
end;
_ -> false
end.
is_loopback({A, _, _, _}) when A =:= 127 -> true;
is_loopback({0, 0, 0, 0, 0, 0, 0, 1}) -> true;
is_loopback(_) -> false.
to_binary(Value) when is_binary(Value) -> Value;
to_binary(Value) when is_list(Value) -> unicode:characters_to_binary(Value);
to_binary(_) -> <<>>.