Current section
Files
Jump to
Current section
Files
src/auth/integrations/openapi/adk_openapi_auth_broker.erl
%% @doc Per-principal OpenAPI credential broker.
%%
%% A broker is normally supervised by the host application and bound to one
%% principal. Its state contains opaque credential references and routing
%% metadata only. API keys and fixed bearer tokens are fetched from a private
%% credential store; OAuth access tokens come from adk_token_manager. Tool
%% arguments and Runner context are never accepted by this process.
-module(adk_openapi_auth_broker).
-behaviour(gen_server).
-behaviour(adk_openapi_auth_manager).
-export([start_link/1, child_spec/1, resolve/2]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3, format_status/1]).
-record(state, {
bindings :: map(),
timeout_ms :: pos_integer(),
max_inflight :: pos_integer(),
inflight = #{} :: map(),
monitors = #{} :: map()
}).
-define(DEFAULT_TIMEOUT_MS, 10000).
-define(MAX_TIMEOUT_MS, 60000).
-define(DEFAULT_MAX_INFLIGHT, 32).
-define(MAX_INFLIGHT, 1024).
-define(CALL_TIMEOUT_MS, 61000).
-define(MAX_OPTIONS_BYTES, 2097152).
-define(MAX_BINDINGS_BYTES, 1048576).
-define(MAX_BINDING_BYTES, 65536).
-define(MAX_BINDINGS, 256).
-define(MAX_REQUEST_BYTES, 65536).
-define(MAX_CREDENTIAL_BYTES, 1048576).
-define(MAX_ID_BYTES, 4096).
-define(MAX_NAME_BYTES, 256).
-define(MAX_PARAMETER_BYTES, 512).
-define(MAX_SCOPES, 64).
-define(MAX_SCOPE_BYTES, 512).
-define(MAX_AUDIENCE_BYTES, 8192).
-define(MAX_API_KEY_BYTES, 16384).
-define(MAX_BEARER_BYTES, 131072).
-define(WORKER_MAX_HEAP_WORDS, 300000).
-spec start_link(map()) -> gen_server:start_ret().
start_link(Opts) when is_map(Opts) ->
case valid_start_options(Opts) of
true ->
case maps:get(name, Opts, undefined) of
undefined -> gen_server:start_link(?MODULE, Opts, []);
Name -> gen_server:start_link({local, Name}, ?MODULE,
Opts, [])
end;
false ->
{error, invalid_openapi_auth_broker_options}
end;
start_link(_Opts) ->
{error, invalid_openapi_auth_broker_options}.
-spec child_spec(map()) -> supervisor:child_spec().
child_spec(Opts) ->
#{id => maps:get(id, Opts, ?MODULE),
start => {?MODULE, start_link, [Opts]},
restart => permanent,
shutdown => 5000,
type => worker,
modules => [?MODULE]}.
-spec resolve(adk_openapi_auth_manager:handle(),
adk_openapi_auth_manager:request()) ->
{ok, adk_openapi_auth_manager:credential()} | {error, term()}.
resolve(Server, Request) when (is_pid(Server) orelse is_atom(Server)),
is_map(Request) ->
%% The broker enforces its own bounded timeout and always replies. The
%% small margin here allows that timeout message to be handled without an
%% unbounded public call.
case bounded_term(Request, ?MAX_REQUEST_BYTES) of
true ->
try gen_server:call(Server, {resolve, Request},
?CALL_TIMEOUT_MS) of
Reply -> Reply
catch
exit:{timeout, _} -> {error, auth_timeout};
exit:_ -> {error, auth_unavailable}
end;
false ->
{error, invalid_auth_request}
end;
resolve(_Server, _Request) ->
{error, invalid_auth_request}.
init(Opts) ->
Bindings0 = maps:get(bindings, Opts, undefined),
Timeout = maps:get(timeout_ms, Opts, ?DEFAULT_TIMEOUT_MS),
MaxInflight = maps:get(max_inflight, Opts, ?DEFAULT_MAX_INFLIGHT),
case {is_integer(Timeout) andalso Timeout > 0 andalso
Timeout =< ?MAX_TIMEOUT_MS,
is_integer(MaxInflight) andalso MaxInflight > 0 andalso
MaxInflight =< ?MAX_INFLIGHT,
normalize_bindings(Bindings0)} of
{true, true, {ok, Bindings}} ->
{ok, #state{bindings = Bindings, timeout_ms = Timeout,
max_inflight = MaxInflight}};
_ ->
{stop, invalid_openapi_auth_broker_options}
end.
handle_call({resolve, Request}, From,
State = #state{bindings = Bindings, timeout_ms = Timeout,
max_inflight = Maximum,
inflight = Inflight, monitors = Monitors}) ->
case prepare_request(Request, Bindings) of
{error, _} = Error ->
{reply, Error, State};
{ok, _Binding} when map_size(Inflight) >= Maximum ->
{reply, {error, auth_capacity_exceeded}, State};
{ok, Binding} ->
JobRef = make_ref(),
Manager = self(),
Deadline = erlang:monotonic_time(millisecond) + Timeout,
ReplyAlias = erlang:alias([explicit_unalias]),
WorkerFun = fun() ->
start_owner_watchdog(Manager, self()),
Result = safe_resolve_binding(Binding, Request, Timeout),
CompletedAt = erlang:monotonic_time(millisecond),
_ = erlang:send(
ReplyAlias,
{openapi_auth_result, JobRef, self(),
CompletedAt, Result},
[noconnect, nosuspend]),
ok
end,
case spawn_auth_worker(WorkerFun) of
{ok, Worker, WorkerMonitor} ->
{Caller, _Tag} = From,
CallerMonitor = erlang:monitor(process, Caller),
Timer = erlang:send_after(
remaining_time(Deadline), self(),
{openapi_auth_timeout, JobRef}),
Entry = #{from => From, worker => Worker,
worker_monitor => WorkerMonitor,
caller_monitor => CallerMonitor, timer => Timer,
reply_alias => ReplyAlias,
deadline => Deadline},
{noreply,
State#state{
inflight = Inflight#{JobRef => Entry},
monitors =
Monitors#{WorkerMonitor => {worker, JobRef},
CallerMonitor => {caller, JobRef}}}};
error ->
_ = erlang:unalias(ReplyAlias),
{reply, {error, auth_unavailable}, State}
end
end;
handle_call(_Request, _From, State) ->
{reply, {error, invalid_auth_request}, State}.
handle_cast(_Message, State) -> {noreply, State}.
handle_info({openapi_auth_result, JobRef, Worker, CompletedAt, Result}, State) ->
case maps:find(JobRef, State#state.inflight) of
{ok, #{worker := Worker, from := From, deadline := Deadline}}
when CompletedAt =< Deadline ->
gen_server:reply(From, Result),
{noreply, remove_job(JobRef, false, State)};
{ok, #{worker := Worker, from := From}} ->
exit(Worker, kill),
gen_server:reply(From, {error, auth_timeout}),
{noreply, remove_job(JobRef, false, State)};
_ ->
{noreply, State}
end;
handle_info({openapi_auth_timeout, JobRef}, State) ->
case maps:find(JobRef, State#state.inflight) of
{ok, #{worker := Worker, from := From}} ->
exit(Worker, kill),
gen_server:reply(From, {error, auth_timeout}),
{noreply, remove_job(JobRef, false, State)};
error ->
{noreply, State}
end;
handle_info({'DOWN', Monitor, process, _Pid, _Reason}, State) ->
case maps:find(Monitor, State#state.monitors) of
{ok, {worker, JobRef}} ->
case maps:find(JobRef, State#state.inflight) of
{ok, #{from := From}} ->
gen_server:reply(From, {error, auth_unavailable}),
{noreply, remove_job(JobRef, false, State)};
error -> {noreply, State}
end;
{ok, {caller, JobRef}} ->
{noreply, remove_job(JobRef, true, State)};
error ->
{noreply, State}
end;
handle_info(_Message, State) -> {noreply, State}.
terminate(_Reason, #state{inflight = Inflight}) ->
maps:foreach(
fun(_JobRef, #{worker := Worker, reply_alias := ReplyAlias}) ->
_ = erlang:unalias(ReplyAlias),
exit(Worker, kill)
end,
Inflight),
ok.
code_change(_OldVsn, State, _Extra) -> {ok, State}.
%% Do not expose principal IDs, provider routing, opaque references, or
%% transient requests through sys:get_status/crash reports.
format_status(Status) ->
maps:map(
fun(state, #state{bindings = Bindings, timeout_ms = Timeout,
max_inflight = Maximum, inflight = Inflight}) ->
#{binding_count => map_size(Bindings), timeout_ms => Timeout,
max_inflight => Maximum,
inflight_count => map_size(Inflight)};
(message, _Message) -> adk_secret_redactor:marker();
(log, _Log) -> [];
(reason, _Reason) -> adk_secret_redactor:marker();
(_Key, Value) -> adk_secret_redactor:redact(Value)
end, Status).
prepare_request(#{scheme_name := Name, scheme_type := Type,
scopes := _Scopes} = Request, Bindings) ->
case valid_auth_request(Request) of
true ->
case maps:find(Name, Bindings) of
{ok, #{kind := Type} = Binding} ->
{ok, Binding};
{ok, _DifferentType} -> {error, scheme_type_mismatch};
error -> {error, unknown_auth_scheme}
end;
false ->
{error, invalid_auth_request}
end;
prepare_request(_Request, _Bindings) ->
{error, invalid_auth_request}.
safe_resolve_binding(Binding, Request, Timeout) ->
try resolve_binding(Binding, Request, Timeout) of
{ok, {Kind, Secret}} = Result
when (Kind =:= api_key orelse Kind =:= bearer),
is_binary(Secret), byte_size(Secret) > 0 ->
case valid_secret(Kind, Secret) of
true -> Result;
false -> {error, credential_unavailable}
end;
{error, Reason} when is_atom(Reason) -> {error, Reason};
_ -> {error, credential_unavailable}
catch
_:_ -> {error, credential_unavailable}
end.
remove_job(JobRef, KillWorker,
State = #state{inflight = Inflight0, monitors = Monitors0}) ->
case maps:take(JobRef, Inflight0) of
{Entry, Inflight} ->
Worker = maps:get(worker, Entry),
_ = erlang:unalias(maps:get(reply_alias, Entry)),
case KillWorker andalso is_process_alive(Worker) of
true -> exit(Worker, kill);
false -> ok
end,
_ = erlang:cancel_timer(maps:get(timer, Entry)),
WorkerMonitor = maps:get(worker_monitor, Entry),
CallerMonitor = maps:get(caller_monitor, Entry),
erlang:demonitor(WorkerMonitor, [flush]),
erlang:demonitor(CallerMonitor, [flush]),
flush_job_result(JobRef, Worker),
State#state{
inflight = Inflight,
monitors = maps:remove(
CallerMonitor,
maps:remove(WorkerMonitor, Monitors0))};
error -> State
end.
spawn_auth_worker(WorkerFun) ->
Options =
[monitor, {message_queue_data, off_heap},
{max_heap_size,
#{size => ?WORKER_MAX_HEAP_WORDS, kill => true,
error_logger => false, include_shared_binaries => true}}],
try spawn_opt(WorkerFun, Options) of
{Worker, Monitor} -> {ok, Worker, Monitor}
catch
_:_ -> error
end.
flush_job_result(JobRef, Worker) ->
receive
{openapi_auth_result, JobRef, Worker, _CompletedAt, _Result} -> ok
after 0 -> ok
end.
remaining_time(Deadline) ->
erlang:max(0, Deadline - erlang:monotonic_time(millisecond)).
start_owner_watchdog(Owner, Worker) ->
Watchdog = fun() -> owner_watchdog(Owner, Worker) end,
_ = spawn_opt(
Watchdog,
[{message_queue_data, off_heap},
{max_heap_size,
#{size => 8192, kill => true, error_logger => false,
include_shared_binaries => true}}]),
ok.
owner_watchdog(Owner, Worker) ->
OwnerMonitor = erlang:monitor(process, Owner),
WorkerMonitor = erlang:monitor(process, Worker),
receive
{'DOWN', OwnerMonitor, process, Owner, _OpaqueReason} ->
exit(Worker, kill),
_ = erlang:demonitor(WorkerMonitor, [flush]),
ok;
{'DOWN', WorkerMonitor, process, Worker, _OpaqueReason} ->
_ = erlang:demonitor(OwnerMonitor, [flush]),
ok
end.
resolve_binding(#{kind := api_key} = Binding, _Request, _Timeout) ->
case fetch_credential(Binding) of
{ok, #{kind := api_key, api_key := Secret}}
when is_binary(Secret), byte_size(Secret) > 0,
byte_size(Secret) =< ?MAX_API_KEY_BYTES ->
{ok, {api_key, Secret}};
_ -> {error, credential_unavailable}
end;
resolve_binding(#{kind := bearer} = Binding, _Request, _Timeout) ->
case fetch_credential(Binding) of
{ok, #{kind := bearer_token, access_token := Secret}}
when is_binary(Secret), byte_size(Secret) > 0,
byte_size(Secret) =< ?MAX_BEARER_BYTES ->
{ok, {bearer, Secret}};
_ -> {error, credential_unavailable}
end;
resolve_binding(#{kind := oauth2, allowed_scopes := Allowed} = Binding,
#{scopes := Scopes}, Timeout) ->
case valid_requested_scopes(Scopes, Allowed) of
false -> {error, scope_not_allowed};
true ->
Request = #{principal => maps:get(principal, Binding),
provider => maps:get(provider, Binding),
credential_ref => maps:get(credential_ref, Binding),
scopes => lists:usort(Scopes),
audience => maps:get(audience, Binding)},
case adk_token_manager:get_token(
maps:get(token_manager, Binding), Request, Timeout) of
{ok, #{access_token := Secret}}
when is_binary(Secret), byte_size(Secret) > 0,
byte_size(Secret) =< ?MAX_BEARER_BYTES ->
{ok, {bearer, Secret}};
_ -> {error, credential_unavailable}
end
end.
fetch_credential(Binding) ->
Module = maps:get(store_module, Binding),
try Module:fetch(maps:get(store_handle, Binding),
maps:get(principal, Binding),
maps:get(provider, Binding),
maps:get(credential_ref, Binding)) of
{ok, Credential} when is_map(Credential) ->
case bounded_term(Credential, ?MAX_CREDENTIAL_BYTES) of
true -> {ok, Credential};
false -> {error, unavailable}
end;
_ -> {error, unavailable}
catch
_:_ -> {error, unavailable}
end.
normalize_bindings(Bindings)
when is_map(Bindings), map_size(Bindings) > 0,
map_size(Bindings) =< ?MAX_BINDINGS ->
case bounded_term(Bindings, ?MAX_BINDINGS_BYTES) of
true -> normalize_binding_pairs(maps:to_list(Bindings), #{});
false -> false
end;
normalize_bindings(_Bindings) -> false.
normalize_binding_pairs([], Acc) -> {ok, Acc};
normalize_binding_pairs([{Name, Binding} | Rest], Acc)
when is_binary(Name), byte_size(Name) > 0,
byte_size(Name) =< ?MAX_NAME_BYTES, is_map(Binding) ->
case normalize_binding(Binding) of
{ok, Normalized} ->
normalize_binding_pairs(Rest, Acc#{Name => Normalized});
error -> false
end;
normalize_binding_pairs(_Pairs, _Acc) -> false.
normalize_binding(#{kind := Kind} = Binding)
when Kind =:= api_key; Kind =:= bearer ->
Allowed = [kind, store_module, store_handle, principal,
provider, credential_ref],
case bounded_term(Binding, ?MAX_BINDING_BYTES) andalso
exact_keys(Binding, Allowed) andalso
valid_store_binding(Binding) of
true -> {ok, Binding};
false -> error
end;
normalize_binding(#{kind := oauth2} = Binding0) ->
Allowed = [kind, token_manager, principal, provider, credential_ref,
allowed_scopes, audience],
Binding = Binding0#{allowed_scopes =>
maps:get(allowed_scopes, Binding0, []),
audience => maps:get(audience, Binding0, undefined)},
case bounded_term(Binding, ?MAX_BINDING_BYTES) andalso
exact_keys(Binding, Allowed) andalso valid_oauth_binding(Binding) of
true -> {ok, Binding};
false -> error
end;
normalize_binding(_Binding) -> error.
valid_store_binding(#{store_module := Module, store_handle := Handle,
principal := Principal, provider := Provider,
credential_ref := Ref}) ->
is_atom(Module) andalso Module =/= undefined andalso
valid_server(Handle) andalso valid_identity(Principal) andalso
valid_identity(Provider) andalso adk_credential_store:is_ref(Ref);
valid_store_binding(_) -> false.
valid_oauth_binding(#{token_manager := Manager, principal := Principal,
provider := Provider, credential_ref := Ref,
allowed_scopes := Scopes, audience := Audience}) ->
valid_server(Manager) andalso valid_identity(Principal) andalso
valid_identity(Provider) andalso adk_credential_store:is_ref(Ref) andalso
valid_scopes(Scopes) andalso valid_audience(Audience);
valid_oauth_binding(_) -> false.
valid_requested_scopes(Scopes, Allowed) ->
valid_scopes(Scopes) andalso
lists:all(fun(Scope) -> lists:member(Scope, Allowed) end, Scopes).
valid_scopes(Scopes) when is_list(Scopes) ->
valid_text_list(Scopes, ?MAX_SCOPES, ?MAX_SCOPE_BYTES);
valid_scopes(_) -> false.
valid_audience(undefined) -> true;
valid_audience(Value) when is_binary(Value) ->
byte_size(Value) > 0 andalso byte_size(Value) =< ?MAX_AUDIENCE_BYTES;
valid_audience(_) -> false.
valid_server(Value) when is_pid(Value) -> true;
valid_server(Value) when is_atom(Value) -> Value =/= undefined;
valid_server(_) -> false.
valid_identity(Value) when is_binary(Value) ->
byte_size(Value) > 0 andalso byte_size(Value) =< ?MAX_ID_BYTES;
valid_identity(Value) when is_atom(Value) -> Value =/= undefined;
valid_identity(_) -> false.
exact_keys(Map, Keys) ->
lists:sort(maps:keys(Map)) =:= lists:sort(Keys).
valid_start_options(Opts) ->
Allowed = [name, id, bindings, timeout_ms, max_inflight],
bounded_term(Opts, ?MAX_OPTIONS_BYTES) andalso
lists:all(fun(Key) -> lists:member(Key, Allowed) end, maps:keys(Opts)) andalso
valid_start_limits(Opts) andalso
case maps:get(name, Opts, undefined) of
undefined -> true;
Name when is_atom(Name) -> Name =/= undefined;
_ -> false
end.
valid_start_limits(Opts) ->
Timeout = maps:get(timeout_ms, Opts, ?DEFAULT_TIMEOUT_MS),
MaxInflight = maps:get(max_inflight, Opts, ?DEFAULT_MAX_INFLIGHT),
is_integer(Timeout) andalso Timeout > 0 andalso
Timeout =< ?MAX_TIMEOUT_MS andalso is_integer(MaxInflight) andalso
MaxInflight > 0 andalso MaxInflight =< ?MAX_INFLIGHT andalso
case normalize_bindings(maps:get(bindings, Opts, undefined)) of
{ok, _} -> true;
_ -> false
end.
valid_auth_request(#{operation_id := OperationId,
scheme_name := Name,
scheme_type := Type,
scopes := Scopes} = Request) ->
Allowed = [operation_id, scheme_name, scheme_type, scopes,
location, parameter_name],
lists:all(fun(Key) -> lists:member(Key, Allowed) end,
maps:keys(Request)) andalso
bounded_term(Request, ?MAX_REQUEST_BYTES) andalso
valid_text(OperationId, ?MAX_NAME_BYTES) andalso
valid_text(Name, ?MAX_NAME_BYTES) andalso
(Type =:= api_key orelse Type =:= bearer orelse Type =:= oauth2) andalso
valid_scopes(Scopes) andalso valid_optional_request_fields(Request);
valid_auth_request(_Request) -> false.
valid_optional_request_fields(Request) ->
ValidLocation = case maps:find(location, Request) of
error -> true;
{ok, Location} -> Location =:= header orelse Location =:= query
end,
ValidParameter = case maps:find(parameter_name, Request) of
error -> true;
{ok, Parameter} -> valid_text(Parameter, ?MAX_PARAMETER_BYTES)
end,
ValidLocation andalso ValidParameter andalso
(maps:is_key(location, Request) =:= maps:is_key(parameter_name, Request)).
valid_secret(api_key, Secret) ->
byte_size(Secret) =< ?MAX_API_KEY_BYTES;
valid_secret(bearer, Secret) ->
byte_size(Secret) =< ?MAX_BEARER_BYTES.
valid_text(Value, Maximum) when is_binary(Value) ->
byte_size(Value) > 0 andalso byte_size(Value) =< Maximum;
valid_text(_Value, _Maximum) -> false.
valid_text_list(List, MaximumCount, MaximumBytes) ->
valid_text_list(List, MaximumCount, MaximumBytes, 0, #{}).
valid_text_list([], _MaximumCount, _MaximumBytes, _Count, _Seen) -> true;
valid_text_list([Value | Rest], MaximumCount, MaximumBytes, Count, Seen)
when Count < MaximumCount, is_binary(Value), byte_size(Value) > 0,
byte_size(Value) =< MaximumBytes ->
case maps:is_key(Value, Seen) of
true -> false;
false -> valid_text_list(Rest, MaximumCount, MaximumBytes,
Count + 1, Seen#{Value => true})
end;
valid_text_list(_List, _MaximumCount, _MaximumBytes, _Count, _Seen) -> false.
bounded_term(Term, Maximum) ->
try erlang:external_size(Term) =< Maximum
catch _:_ -> false
end.