mirror of
https://github.com/fluxerapp/fluxer
synced 2026-10-08 03:32:27 +09:00
perf(push): cache endpoint guard dns verdicts (#2907)
This commit is contained in:
@@ -81,6 +81,9 @@ env_gateway_base_config() ->
|
||||
<<"push_clear_notifications_enabled">> => env_bool(
|
||||
"FLUXER_GATEWAY_PUSH_CLEAR_NOTIFICATIONS_ENABLED", true
|
||||
),
|
||||
<<"push_endpoint_guard_enabled">> => env_bool(
|
||||
"FLUXER_GATEWAY_PUSH_ENDPOINT_GUARD_ENABLED", true
|
||||
),
|
||||
<<"push_outbox_request_timeout_ms">> => env_int(
|
||||
"FLUXER_GATEWAY_PUSH_OUTBOX_REQUEST_TIMEOUT_MS", 100000
|
||||
),
|
||||
@@ -256,6 +259,9 @@ build_push_config(Service, Public) ->
|
||||
push_clear_notifications_enabled => get_bool(
|
||||
Service, <<"push_clear_notifications_enabled">>, true
|
||||
),
|
||||
push_endpoint_guard_enabled => get_bool(
|
||||
Service, <<"push_endpoint_guard_enabled">>, true
|
||||
),
|
||||
push_outbox_max_queue => get_int(Service, <<"push_outbox_max_queue">>, 10000),
|
||||
push_outbox_max_inflight => get_int(Service, <<"push_outbox_max_inflight">>, 64),
|
||||
push_outbox_request_timeout_ms => get_int(
|
||||
|
||||
@@ -3,24 +3,46 @@
|
||||
-module(push_endpoint_guard).
|
||||
-typing([eqwalizer]).
|
||||
|
||||
-export([check/1, check/2]).
|
||||
-export([check/1, check/2, check/3]).
|
||||
|
||||
-export_type([resolver/0]).
|
||||
-export_type([resolver/0, cache_mode/0, verdict/0]).
|
||||
|
||||
-define(RESOLVE_TIMEOUT_MS, 3000).
|
||||
-define(MAX_HOST_LENGTH, 253).
|
||||
-define(MAX_LABEL_LENGTH, 63).
|
||||
-define(ALLOWED_VERDICT_TTL_SECONDS, 300).
|
||||
-define(REFUSED_VERDICT_TTL_SECONDS, 30).
|
||||
|
||||
-type resolver() :: fun((string()) -> {ok, [inet:ip_address()]} | {error, term()}).
|
||||
-type cache_mode() :: cached | uncached.
|
||||
-type verdict() :: ok | {error, term()}.
|
||||
|
||||
-spec check(binary()) -> ok | {error, term()}.
|
||||
-spec check(binary()) -> verdict().
|
||||
check(Endpoint) ->
|
||||
check(Endpoint, fun resolve/1).
|
||||
check(Endpoint, fun resolve/1, cached).
|
||||
|
||||
-spec check(binary(), resolver()) -> ok | {error, term()}.
|
||||
-spec check(binary(), resolver()) -> verdict().
|
||||
check(Endpoint, Resolver) ->
|
||||
check(Endpoint, Resolver, uncached).
|
||||
|
||||
-spec check(binary(), resolver(), cache_mode()) -> verdict().
|
||||
check(Endpoint, Resolver, CacheMode) ->
|
||||
case enabled() of
|
||||
true -> check_endpoint(Endpoint, Resolver, CacheMode);
|
||||
false -> ok
|
||||
end.
|
||||
|
||||
-spec enabled() -> boolean().
|
||||
enabled() ->
|
||||
case fluxer_gateway_env:get(push_endpoint_guard_enabled) of
|
||||
Enabled when is_boolean(Enabled) -> Enabled;
|
||||
_ -> true
|
||||
end.
|
||||
|
||||
-spec check_endpoint(binary(), resolver(), cache_mode()) -> verdict().
|
||||
check_endpoint(Endpoint, Resolver, CacheMode) ->
|
||||
case parse_endpoint(Endpoint) of
|
||||
{ok, Host} -> check_host(Host, Resolver);
|
||||
{ok, Host} -> check_host(Host, Resolver, CacheMode);
|
||||
{error, Reason} -> {error, Reason}
|
||||
end.
|
||||
|
||||
@@ -64,21 +86,41 @@ allowed_port(80) -> true;
|
||||
allowed_port(443) -> true;
|
||||
allowed_port(_Port) -> false.
|
||||
|
||||
-spec check_host(string(), resolver()) -> ok | {error, term()}.
|
||||
check_host(Host, Resolver) ->
|
||||
-spec check_host(string(), resolver(), cache_mode()) -> verdict().
|
||||
check_host(Host, Resolver, CacheMode) ->
|
||||
case inet:parse_address(Host) of
|
||||
{ok, Address} -> check_addresses([Address]);
|
||||
{error, _Reason} -> check_hostname(Host, Resolver)
|
||||
{error, _Reason} -> check_hostname(Host, Resolver, CacheMode)
|
||||
end.
|
||||
|
||||
-spec check_hostname(string(), resolver()) -> ok | {error, term()}.
|
||||
check_hostname(Host, Resolver) ->
|
||||
-spec check_hostname(string(), resolver(), cache_mode()) -> verdict().
|
||||
check_hostname(Host, Resolver, CacheMode) ->
|
||||
case is_fqdn(Host) of
|
||||
true -> resolve_and_check(Host, Resolver);
|
||||
true -> resolve_and_check(Host, Resolver, CacheMode);
|
||||
false -> {error, endpoint_rejected}
|
||||
end.
|
||||
|
||||
-spec resolve_and_check(string(), resolver()) -> ok | {error, term()}.
|
||||
-spec resolve_and_check(string(), resolver(), cache_mode()) -> verdict().
|
||||
resolve_and_check(Host, Resolver, uncached) ->
|
||||
resolve_and_check(Host, Resolver);
|
||||
resolve_and_check(Host, Resolver, cached) ->
|
||||
CacheKey = list_to_binary(Host),
|
||||
case push_ets_cache:get_endpoint_verdict(CacheKey) of
|
||||
{ok, Verdict} -> Verdict;
|
||||
undefined -> store_verdict(CacheKey, resolve_and_check(Host, Resolver))
|
||||
end.
|
||||
|
||||
-spec store_verdict(binary(), verdict()) -> verdict().
|
||||
store_verdict(CacheKey, Verdict) ->
|
||||
Ttl = verdict_ttl_seconds(Verdict),
|
||||
ok = push_ets_cache:put_endpoint_verdict(CacheKey, Verdict, Ttl),
|
||||
Verdict.
|
||||
|
||||
-spec verdict_ttl_seconds(verdict()) -> pos_integer().
|
||||
verdict_ttl_seconds(ok) -> ?ALLOWED_VERDICT_TTL_SECONDS;
|
||||
verdict_ttl_seconds({error, _Reason}) -> ?REFUSED_VERDICT_TTL_SECONDS.
|
||||
|
||||
-spec resolve_and_check(string(), resolver()) -> verdict().
|
||||
resolve_and_check(Host, Resolver) ->
|
||||
case Resolver(Host) of
|
||||
{ok, Addresses} -> check_addresses(Addresses);
|
||||
|
||||
@@ -25,6 +25,8 @@
|
||||
reserve_badge_counts/1,
|
||||
get_bearer_token/1,
|
||||
put_bearer_token/3,
|
||||
get_endpoint_verdict/1,
|
||||
put_endpoint_verdict/3,
|
||||
release/1,
|
||||
rebalance/0,
|
||||
rebalance_async/0,
|
||||
@@ -33,16 +35,20 @@
|
||||
table_size/1
|
||||
]).
|
||||
|
||||
-export_type([fill/0]).
|
||||
-export_type([fill/0, endpoint_verdict/0]).
|
||||
|
||||
-define(USER_GUILD_SETTINGS, push_user_guild_settings).
|
||||
-define(SUBSCRIPTIONS, push_subscriptions).
|
||||
-define(BLOCKED_IDS, push_blocked_ids).
|
||||
-define(BADGE_COUNTS, push_badge_counts).
|
||||
-define(BEARER_TOKENS, push_bearer_tokens).
|
||||
-define(ENDPOINT_VERDICTS, push_endpoint_verdicts).
|
||||
|
||||
-define(MAX_TABLE_ENTRIES, 500000).
|
||||
-define(MAX_BEARER_TOKENS, 10000).
|
||||
-define(MAX_ENDPOINT_VERDICTS, 2048).
|
||||
-define(MAX_ENDPOINT_HOST_BYTES, 253).
|
||||
-define(ENDPOINT_VERDICT_EVICT_BATCH, 512).
|
||||
-define(EVICT_BATCH, 4096).
|
||||
-define(MAX_EVICT_RESEEKS, 8).
|
||||
-define(RESERVATION_TTL_MS, 120000).
|
||||
@@ -55,6 +61,7 @@
|
||||
]).
|
||||
|
||||
-type fill() :: {atom(), pos_integer(), [term()]}.
|
||||
-type endpoint_verdict() :: ok | {error, term()}.
|
||||
|
||||
-spec init() -> ok.
|
||||
init() ->
|
||||
@@ -63,6 +70,7 @@ init() ->
|
||||
ensure_table(?BLOCKED_IDS),
|
||||
ensure_table(?BADGE_COUNTS),
|
||||
ensure_table(?BEARER_TOKENS),
|
||||
ensure_table(?ENDPOINT_VERDICTS),
|
||||
ok.
|
||||
|
||||
-spec get_user_guild_settings(integer(), integer()) -> map() | undefined.
|
||||
@@ -219,6 +227,48 @@ put_bearer_token(Key, Token, ExpiresAt) when is_binary(Token), is_integer(Expire
|
||||
error:badarg -> ok
|
||||
end.
|
||||
|
||||
-spec get_endpoint_verdict(binary()) -> {ok, endpoint_verdict()} | undefined.
|
||||
get_endpoint_verdict(Host) when is_binary(Host) ->
|
||||
try ets:lookup(?ENDPOINT_VERDICTS, Host) of
|
||||
[{_, ok, ExpiresAt}] when is_integer(ExpiresAt) ->
|
||||
live_endpoint_verdict(ok, ExpiresAt);
|
||||
[{_, {error, Reason}, ExpiresAt}] when is_integer(ExpiresAt) ->
|
||||
live_endpoint_verdict({error, Reason}, ExpiresAt);
|
||||
_ ->
|
||||
undefined
|
||||
catch
|
||||
error:badarg -> undefined
|
||||
end.
|
||||
|
||||
-spec live_endpoint_verdict(endpoint_verdict(), integer()) ->
|
||||
{ok, endpoint_verdict()} | undefined.
|
||||
live_endpoint_verdict(Verdict, ExpiresAt) ->
|
||||
case erlang:system_time(second) < ExpiresAt of
|
||||
true -> {ok, Verdict};
|
||||
false -> undefined
|
||||
end.
|
||||
|
||||
-spec put_endpoint_verdict(binary(), endpoint_verdict(), pos_integer()) -> ok.
|
||||
put_endpoint_verdict(Host, Verdict, TtlSeconds) when
|
||||
is_binary(Host), is_integer(TtlSeconds), TtlSeconds > 0
|
||||
->
|
||||
case byte_size(Host) =< ?MAX_ENDPOINT_HOST_BYTES of
|
||||
true -> insert_endpoint_verdict(Host, Verdict, TtlSeconds);
|
||||
false -> ok
|
||||
end.
|
||||
|
||||
-spec insert_endpoint_verdict(binary(), endpoint_verdict(), pos_integer()) -> ok.
|
||||
insert_endpoint_verdict(Host, Verdict, TtlSeconds) ->
|
||||
guard_table_size(
|
||||
?ENDPOINT_VERDICTS, ?MAX_ENDPOINT_VERDICTS, ?ENDPOINT_VERDICT_EVICT_BATCH
|
||||
),
|
||||
ExpiresAt = erlang:system_time(second) + TtlSeconds,
|
||||
try ets:insert(?ENDPOINT_VERDICTS, {Host, Verdict, ExpiresAt}) of
|
||||
_ -> ok
|
||||
catch
|
||||
error:badarg -> ok
|
||||
end.
|
||||
|
||||
-spec write(atom(), tuple()) -> ok.
|
||||
write(Table, Row) ->
|
||||
guard_table_size(Table, ?MAX_TABLE_ENTRIES),
|
||||
@@ -292,7 +342,8 @@ cache_stats() ->
|
||||
push_subscriptions_size => table_size(?SUBSCRIPTIONS),
|
||||
blocked_ids_size => table_size(?BLOCKED_IDS),
|
||||
badge_counts_size => table_size(?BADGE_COUNTS),
|
||||
bearer_tokens_size => table_size(?BEARER_TOKENS)
|
||||
bearer_tokens_size => table_size(?BEARER_TOKENS),
|
||||
endpoint_verdicts_size => table_size(?ENDPOINT_VERDICTS)
|
||||
}.
|
||||
|
||||
-spec evict_tables(map()) -> ok.
|
||||
@@ -300,6 +351,7 @@ evict_tables(MaxEntries) ->
|
||||
Now = erlang:system_time(second),
|
||||
select_delete(?BLOCKED_IDS, expired_rows(Now)),
|
||||
select_delete(?BEARER_TOKENS, expired_rows(Now)),
|
||||
select_delete(?ENDPOINT_VERDICTS, expired_rows(Now)),
|
||||
lists:foreach(
|
||||
fun expire_reservations/1,
|
||||
[?USER_GUILD_SETTINGS, ?SUBSCRIPTIONS, ?BLOCKED_IDS, ?BADGE_COUNTS]
|
||||
@@ -309,6 +361,7 @@ evict_tables(MaxEntries) ->
|
||||
evict_table(?BLOCKED_IDS, maps:get(blocked_ids, MaxEntries, undefined)),
|
||||
evict_table(?BADGE_COUNTS, maps:get(badge_counts, MaxEntries, undefined)),
|
||||
evict_table(?BEARER_TOKENS, ?MAX_BEARER_TOKENS),
|
||||
evict_table(?ENDPOINT_VERDICTS, ?MAX_ENDPOINT_VERDICTS),
|
||||
ok.
|
||||
|
||||
-spec expired_rows(integer()) -> ets:match_spec().
|
||||
@@ -330,8 +383,12 @@ select_delete(Table, MatchSpec) ->
|
||||
|
||||
-spec guard_table_size(atom(), non_neg_integer()) -> ok.
|
||||
guard_table_size(Table, MaxEntries) ->
|
||||
guard_table_size(Table, MaxEntries, ?EVICT_BATCH).
|
||||
|
||||
-spec guard_table_size(atom(), non_neg_integer(), pos_integer()) -> ok.
|
||||
guard_table_size(Table, MaxEntries, EvictBatch) ->
|
||||
case table_size(Table) >= MaxEntries of
|
||||
true -> evict_table(Table, max(0, MaxEntries - ?EVICT_BATCH));
|
||||
true -> evict_table(Table, max(0, MaxEntries - EvictBatch));
|
||||
false -> ok
|
||||
end.
|
||||
|
||||
|
||||
@@ -80,6 +80,16 @@ cluster_static_peers_accepts_valid_node_names_test() ->
|
||||
maps:get(cluster_static_peers, Config)
|
||||
).
|
||||
|
||||
push_endpoint_guard_enabled_defaults_on_test() ->
|
||||
Config = fluxer_gateway_config:load(),
|
||||
?assertEqual(true, maps:get(push_endpoint_guard_enabled, Config)).
|
||||
|
||||
push_endpoint_guard_enabled_can_be_turned_off_test() ->
|
||||
with_env("FLUXER_GATEWAY_PUSH_ENDPOINT_GUARD_ENABLED", "false", fun() ->
|
||||
Config = fluxer_gateway_config:load(),
|
||||
?assertEqual(false, maps:get(push_endpoint_guard_enabled, Config))
|
||||
end).
|
||||
|
||||
presence_push_buffer_env_defaults_test() ->
|
||||
with_env("FLUXER_GATEWAY_PRESENCE_PUSH_BUFFER_MAX_ENTRIES", "7", fun() ->
|
||||
with_env("FLUXER_GATEWAY_PRESENCE_PUSH_BUFFER_MAX_BYTES", "4096", fun() ->
|
||||
|
||||
@@ -6,6 +6,8 @@
|
||||
-include_lib("eunit/include/eunit.hrl").
|
||||
|
||||
-define(ENDPOINT, <<"https://push.example.com/wpush/v2/abc">>).
|
||||
-define(VERDICT_TABLE, push_endpoint_verdicts).
|
||||
-define(MAX_VERDICTS, 2048).
|
||||
|
||||
resolves_to(Addresses) ->
|
||||
fun(_Host) -> {ok, Addresses} end.
|
||||
@@ -13,6 +15,9 @@ resolves_to(Addresses) ->
|
||||
fails_with(Reason) ->
|
||||
fun(_Host) -> {error, Reason} end.
|
||||
|
||||
never_resolves() ->
|
||||
fun(_Host) -> erlang:error(resolver_called) end.
|
||||
|
||||
link_local_metadata_address_is_refused_test() ->
|
||||
?assertEqual(
|
||||
{error, endpoint_blocked},
|
||||
@@ -185,8 +190,162 @@ userinfo_is_refused_test() ->
|
||||
)
|
||||
).
|
||||
|
||||
a_repeat_lookup_for_the_same_host_does_not_resolve_again_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
Counter = counters:new(1, []),
|
||||
Endpoint = endpoint("cache-repeat.example.com"),
|
||||
Resolver = counting_resolver(Counter, [{93, 184, 216, 34}]),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(1, counters:get(Counter, 1))
|
||||
end).
|
||||
|
||||
a_cache_hit_returns_the_verdict_the_uncached_path_returns_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
lists:foreach(
|
||||
fun assert_cached_matches_uncached/1,
|
||||
[
|
||||
{"cache-allow.example.com", resolves_to([{93, 184, 216, 34}]), ok},
|
||||
{"cache-block.example.com", resolves_to([{10, 0, 0, 1}]),
|
||||
{error, endpoint_blocked}},
|
||||
{"cache-empty.example.com", resolves_to([]), {error, nxdomain}},
|
||||
{"cache-timeout.example.com", fails_with(timeout), {error, timeout}}
|
||||
]
|
||||
)
|
||||
end).
|
||||
|
||||
assert_cached_matches_uncached({Host, Resolver, Expected}) ->
|
||||
Endpoint = endpoint(Host),
|
||||
?assertEqual(Expected, push_endpoint_guard:check(Endpoint, Resolver)),
|
||||
?assertEqual(Expected, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(Expected, push_endpoint_guard:check(Endpoint, never_resolves(), cached)).
|
||||
|
||||
an_uncached_check_never_writes_the_cache_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
Endpoint = endpoint("cache-bypass.example.com"),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, resolves_to([{1, 1, 1, 1}]))),
|
||||
?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE))
|
||||
end).
|
||||
|
||||
an_ip_literal_is_never_cached_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
?assertEqual(
|
||||
ok, push_endpoint_guard:check(<<"https://93.184.216.34/sub">>, never_resolves())
|
||||
),
|
||||
?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE))
|
||||
end).
|
||||
|
||||
a_cached_verdict_expires_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
Counter = counters:new(1, []),
|
||||
Endpoint = endpoint("cache-expiry.example.com"),
|
||||
Resolver = counting_resolver(Counter, [{93, 184, 216, 34}]),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(1, counters:get(Counter, 1)),
|
||||
expire_verdict(<<"cache-expiry.example.com">>),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assertEqual(2, counters:get(Counter, 1))
|
||||
end).
|
||||
|
||||
a_refused_verdict_expires_sooner_than_an_allowed_one_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
Allowed = endpoint("cache-ttl-allowed.example.com"),
|
||||
Refused = endpoint("cache-ttl-refused.example.com"),
|
||||
?assertEqual(
|
||||
ok, push_endpoint_guard:check(Allowed, resolves_to([{93, 184, 216, 34}]), cached)
|
||||
),
|
||||
?assertEqual(
|
||||
{error, timeout}, push_endpoint_guard:check(Refused, fails_with(timeout), cached)
|
||||
),
|
||||
AllowedExpiry = expires_at(<<"cache-ttl-allowed.example.com">>),
|
||||
RefusedExpiry = expires_at(<<"cache-ttl-refused.example.com">>),
|
||||
?assert(RefusedExpiry < AllowedExpiry)
|
||||
end).
|
||||
|
||||
many_distinct_hosts_cannot_grow_the_cache_without_bound_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
Resolver = resolves_to([{93, 184, 216, 34}]),
|
||||
lists:foreach(
|
||||
fun(N) -> flood_one_host(N, Resolver) end,
|
||||
lists:seq(1, 20000)
|
||||
),
|
||||
Size = push_ets_cache:table_size(?VERDICT_TABLE),
|
||||
?assert(Size >= 1500),
|
||||
?assert(Size =< ?MAX_VERDICTS),
|
||||
?assert(verdict_table_bytes() =< 4 * 1024 * 1024)
|
||||
end).
|
||||
|
||||
flood_one_host(N, Resolver) ->
|
||||
Endpoint = endpoint("flood-" ++ integer_to_list(N) ++ ".example.com"),
|
||||
?assertEqual(ok, push_endpoint_guard:check(Endpoint, Resolver, cached)),
|
||||
?assert(push_ets_cache:table_size(?VERDICT_TABLE) =< ?MAX_VERDICTS).
|
||||
|
||||
the_guard_is_enabled_by_default_test() ->
|
||||
?assertEqual(
|
||||
{error, endpoint_blocked},
|
||||
push_endpoint_guard:check(?ENDPOINT, resolves_to([{10, 0, 0, 1}]))
|
||||
).
|
||||
|
||||
a_disabled_guard_passes_everything_through_without_resolving_test() ->
|
||||
with_verdict_cache(fun() ->
|
||||
with_guard_disabled(fun() ->
|
||||
Never = never_resolves(),
|
||||
?assertEqual(ok, push_endpoint_guard:check(?ENDPOINT, Never)),
|
||||
?assertEqual(ok, push_endpoint_guard:check(?ENDPOINT, Never, cached)),
|
||||
?assertEqual(
|
||||
ok, push_endpoint_guard:check(<<"https://169.254.169.254/latest">>, Never)
|
||||
),
|
||||
?assertEqual(
|
||||
ok, push_endpoint_guard:check(<<"http://push.example.com/sub">>, Never)
|
||||
),
|
||||
?assertEqual(ok, push_endpoint_guard:check(<<"not-a-url">>, Never)),
|
||||
?assertEqual(0, push_ets_cache:table_size(?VERDICT_TABLE))
|
||||
end)
|
||||
end).
|
||||
|
||||
counting_resolver(Counter, Addresses) ->
|
||||
fun(_Host) ->
|
||||
counters:add(Counter, 1, 1),
|
||||
{ok, Addresses}
|
||||
end.
|
||||
|
||||
endpoint(Host) ->
|
||||
list_to_binary("https://" ++ Host ++ "/sub").
|
||||
|
||||
expires_at(Host) ->
|
||||
[{Host, _Verdict, ExpiresAt}] = ets:lookup(?VERDICT_TABLE, Host),
|
||||
ExpiresAt.
|
||||
|
||||
expire_verdict(Host) ->
|
||||
[{Host, Verdict, _ExpiresAt}] = ets:lookup(?VERDICT_TABLE, Host),
|
||||
true = ets:insert(?VERDICT_TABLE, {Host, Verdict, erlang:system_time(second) - 1}),
|
||||
ok.
|
||||
|
||||
verdict_table_bytes() ->
|
||||
ets:info(?VERDICT_TABLE, memory) * erlang:system_info(wordsize).
|
||||
|
||||
with_verdict_cache(Fun) ->
|
||||
ok = push_ets_cache:init(),
|
||||
true = ets:delete_all_objects(?VERDICT_TABLE),
|
||||
try
|
||||
Fun()
|
||||
after
|
||||
ets:delete_all_objects(?VERDICT_TABLE)
|
||||
end.
|
||||
|
||||
with_guard_disabled(Fun) ->
|
||||
Original = fluxer_gateway_env:get(push_endpoint_guard_enabled),
|
||||
_ = fluxer_gateway_env:patch(#{push_endpoint_guard_enabled => false}),
|
||||
try
|
||||
Fun()
|
||||
after
|
||||
_ = fluxer_gateway_env:patch(#{push_endpoint_guard_enabled => Original})
|
||||
end.
|
||||
|
||||
ip_literals_skip_dns_and_are_screened_directly_test() ->
|
||||
Never = fun(_Host) -> erlang:error(resolver_called) end,
|
||||
Never = never_resolves(),
|
||||
?assertEqual(
|
||||
{error, endpoint_blocked},
|
||||
push_endpoint_guard:check(<<"https://169.254.169.254/latest">>, Never)
|
||||
@@ -204,7 +363,7 @@ ip_literals_skip_dns_and_are_screened_directly_test() ->
|
||||
?assertEqual(ok, push_endpoint_guard:check(<<"https://93.184.216.34/sub">>, Never)).
|
||||
|
||||
malformed_and_non_fqdn_hosts_are_refused_test() ->
|
||||
Never = fun(_Host) -> erlang:error(resolver_called) end,
|
||||
Never = never_resolves(),
|
||||
lists:foreach(
|
||||
fun(Endpoint) ->
|
||||
?assertEqual(
|
||||
|
||||
@@ -11,6 +11,7 @@ init_creates_tables_test() ->
|
||||
?assertNotEqual(undefined, ets:whereis(push_subscriptions)),
|
||||
?assertNotEqual(undefined, ets:whereis(push_blocked_ids)),
|
||||
?assertNotEqual(undefined, ets:whereis(push_badge_counts)),
|
||||
?assertNotEqual(undefined, ets:whereis(push_endpoint_verdicts)),
|
||||
cleanup_tables().
|
||||
|
||||
init_idempotent_test() ->
|
||||
@@ -112,6 +113,69 @@ rebalance_evicts_remote_owned_entries_test() ->
|
||||
persistent_term:erase({gateway_cluster_membership, members_by_role}),
|
||||
cleanup_tables().
|
||||
|
||||
endpoint_verdict_round_trip_test() ->
|
||||
cleanup_tables(),
|
||||
ok = push_ets_cache:init(),
|
||||
?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(<<"push.example.com">>)),
|
||||
ok = push_ets_cache:put_endpoint_verdict(<<"push.example.com">>, ok, 300),
|
||||
?assertEqual({ok, ok}, push_ets_cache:get_endpoint_verdict(<<"push.example.com">>)),
|
||||
ok = push_ets_cache:put_endpoint_verdict(
|
||||
<<"bad.example.com">>, {error, endpoint_blocked}, 30
|
||||
),
|
||||
?assertEqual(
|
||||
{ok, {error, endpoint_blocked}},
|
||||
push_ets_cache:get_endpoint_verdict(<<"bad.example.com">>)
|
||||
),
|
||||
cleanup_tables().
|
||||
|
||||
endpoint_verdicts_expire_and_are_reclaimed_test() ->
|
||||
cleanup_tables(),
|
||||
ok = push_ets_cache:init(),
|
||||
ok = push_ets_cache:put_endpoint_verdict(<<"stale.example.com">>, ok, 300),
|
||||
Stale = erlang:system_time(second) - 1,
|
||||
true = ets:insert(push_endpoint_verdicts, {<<"stale.example.com">>, ok, Stale}),
|
||||
?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(<<"stale.example.com">>)),
|
||||
ok = push_ets_cache:evict_tables(#{}),
|
||||
?assertEqual([], ets:lookup(push_endpoint_verdicts, <<"stale.example.com">>)),
|
||||
cleanup_tables().
|
||||
|
||||
an_oversized_host_is_never_cached_test() ->
|
||||
cleanup_tables(),
|
||||
ok = push_ets_cache:init(),
|
||||
Oversized = binary:copy(<<"a">>, 254),
|
||||
ok = push_ets_cache:put_endpoint_verdict(Oversized, ok, 300),
|
||||
?assertEqual(undefined, push_ets_cache:get_endpoint_verdict(Oversized)),
|
||||
?assertEqual(0, push_ets_cache:table_size(push_endpoint_verdicts)),
|
||||
AtLimit = binary:copy(<<"a">>, 253),
|
||||
ok = push_ets_cache:put_endpoint_verdict(AtLimit, ok, 300),
|
||||
?assertEqual({ok, ok}, push_ets_cache:get_endpoint_verdict(AtLimit)),
|
||||
cleanup_tables().
|
||||
|
||||
endpoint_verdicts_stay_bounded_under_max_length_hosts_test() ->
|
||||
cleanup_tables(),
|
||||
ok = push_ets_cache:init(),
|
||||
lists:foreach(fun seed_max_length_verdict/1, lists:seq(1, 20000)),
|
||||
Size = push_ets_cache:table_size(push_endpoint_verdicts),
|
||||
?assert(Size >= 1500),
|
||||
?assert(Size =< 2048),
|
||||
Bytes = ets:info(push_endpoint_verdicts, memory) * erlang:system_info(wordsize),
|
||||
?assert(Bytes =< 4 * 1024 * 1024),
|
||||
cleanup_tables().
|
||||
|
||||
seed_max_length_verdict(N) ->
|
||||
Suffix = integer_to_binary(N),
|
||||
Host = <<(binary:copy(<<"a">>, 253 - byte_size(Suffix)))/binary, Suffix/binary>>,
|
||||
ok = push_ets_cache:put_endpoint_verdict(Host, ok, 300),
|
||||
?assert(push_ets_cache:table_size(push_endpoint_verdicts) =< 2048).
|
||||
|
||||
endpoint_verdicts_are_reported_in_cache_stats_test() ->
|
||||
cleanup_tables(),
|
||||
ok = push_ets_cache:init(),
|
||||
ok = push_ets_cache:put_endpoint_verdict(<<"push.example.com">>, ok, 300),
|
||||
Stats = push_ets_cache:cache_stats(),
|
||||
?assertEqual(1, maps:get(endpoint_verdicts_size, Stats)),
|
||||
cleanup_tables().
|
||||
|
||||
seed_subscriptions(UserId, Subscriptions) ->
|
||||
push_ets_cache:put_subscriptions(
|
||||
UserId, Subscriptions, push_ets_cache:reserve_subscriptions([UserId])
|
||||
@@ -143,6 +207,7 @@ cleanup_tables() ->
|
||||
delete_table(push_blocked_ids),
|
||||
delete_table(push_badge_counts),
|
||||
delete_table(push_bearer_tokens),
|
||||
delete_table(push_endpoint_verdicts),
|
||||
ok.
|
||||
|
||||
delete_table(Table) ->
|
||||
|
||||
Reference in New Issue
Block a user