feat(mobile): rate-limit socket + batch historical packet delivery
- MobileUserSocket: cap new socket connections per IP at 30/min via Aprsme.RateLimiter. Denials log and return :error at handshake. Tests bypass the check (no real peer_data). - MobileChannel: rate-limit the expensive client-initiated handlers (subscribe_bounds, subscribe_callsign, search_callsign) at 30/minute per socket. Streaming packet delivery is unaffected. - MobileChannel: historical packet loads now arrive as a batched `packets` event (100 per frame) instead of one `packet` frame per row. Live streaming still uses the single `packet` event. - Updated docs/mobile-api.md with the `packets` event and rate-limit semantics; migration note preserves compatibility for clients that only handle live `packet` events.
This commit is contained in:
parent
8f64d0e999
commit
09f92c6738
4 changed files with 187 additions and 80 deletions
|
|
@ -189,6 +189,46 @@ Once subscribed, you'll receive packets within your bounds. Historical packets a
|
|||
|
||||
**Note:** Optional fields are omitted from the payload entirely when their value is `nil`.
|
||||
|
||||
### Receiving Batched Packets (Historical Loads)
|
||||
|
||||
When the server delivers historical packets in bulk (for example, the replay
|
||||
that follows `subscribe_bounds` or `subscribe_callsign`), they arrive as a
|
||||
batched `packets` event instead of many individual `packet` events. Live
|
||||
streaming packets continue to use the single `packet` event described above.
|
||||
|
||||
**Event:** `packets`
|
||||
|
||||
**Payload:**
|
||||
```json
|
||||
{
|
||||
"packets": [
|
||||
{ "id": "...", "callsign": "...", "lat": 33.1, "lng": -96.1, "timestamp": "2025-10-25T16:17:20Z", "symbol_table_id": "/", "symbol_code": ">" },
|
||||
{ "id": "...", "callsign": "...", "lat": 33.2, "lng": -96.2, "timestamp": "2025-10-25T16:17:25Z", "symbol_table_id": "/", "symbol_code": "#" }
|
||||
],
|
||||
"count": 2
|
||||
}
|
||||
```
|
||||
|
||||
Each entry inside `packets` has the same schema as the single-packet `packet`
|
||||
event above. Batch sizes are capped at 100 entries per frame, so a 1,000-packet
|
||||
history load arrives as ~10 `packets` events.
|
||||
|
||||
**Migration guidance:** existing clients that only listen for the `packet`
|
||||
event will still receive live updates correctly — they simply won't populate
|
||||
the historical backlog. To show history, add a handler for `packets` that
|
||||
iterates over `payload.packets` and applies the same rendering logic.
|
||||
|
||||
### Rate Limiting
|
||||
|
||||
- **Connect:** up to 30 new socket connections per IP per minute. Exceeding the
|
||||
limit returns an immediate `:error` from the socket handshake; retry after
|
||||
the minute window elapses.
|
||||
- **Channel requests:** the expensive operations (`subscribe_bounds`,
|
||||
`subscribe_callsign`, `search_callsign`) are rate-limited to 30 calls per
|
||||
socket per minute. Exceeding the limit returns
|
||||
`{ "status": "error", "response": { "message": "Rate limit exceeded; retry later", "retry_after_ms": <ms> } }`.
|
||||
- Streaming packet delivery is *not* rate-limited.
|
||||
|
||||
### Unsubscribing
|
||||
|
||||
Stop receiving packets.
|
||||
|
|
|
|||
|
|
@ -57,6 +57,11 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
|
||||
require Logger
|
||||
|
||||
# Per-socket rate limits on the expensive operations. Streaming packet
|
||||
# delivery is unaffected — these only guard client-initiated queries.
|
||||
@expensive_op_limit 30
|
||||
@expensive_op_scale 60_000
|
||||
|
||||
@impl true
|
||||
def join("mobile:packets", _payload, socket) do
|
||||
Logger.info("Mobile client joined packets channel")
|
||||
|
|
@ -71,38 +76,40 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
) do
|
||||
Logger.info("Mobile websocket received subscribe_bounds: #{inspect(payload)}")
|
||||
|
||||
bounds = %{
|
||||
north: ensure_float(north),
|
||||
south: ensure_float(south),
|
||||
east: ensure_float(east),
|
||||
west: ensure_float(west)
|
||||
}
|
||||
with :ok <- check_rate_limit(socket, "subscribe_bounds") do
|
||||
bounds = %{
|
||||
north: ensure_float(north),
|
||||
south: ensure_float(south),
|
||||
east: ensure_float(east),
|
||||
west: ensure_float(west)
|
||||
}
|
||||
|
||||
# Validate bounds
|
||||
case validate_bounds(bounds) do
|
||||
:ok ->
|
||||
# Generate unique client ID
|
||||
client_id = "mobile_#{:erlang.phash2(self())}"
|
||||
# Validate bounds
|
||||
case validate_bounds(bounds) do
|
||||
:ok ->
|
||||
# Generate unique client ID
|
||||
client_id = "mobile_#{:erlang.phash2(self())}"
|
||||
|
||||
# Subscribe to StreamingPacketsPubSub with bounds
|
||||
Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), bounds)
|
||||
# Subscribe to StreamingPacketsPubSub with bounds
|
||||
Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), bounds)
|
||||
|
||||
# Store client info in socket
|
||||
socket =
|
||||
socket
|
||||
|> assign(:client_id, client_id)
|
||||
|> assign(:bounds, bounds)
|
||||
|> assign(:subscribed, true)
|
||||
# Store client info in socket
|
||||
socket =
|
||||
socket
|
||||
|> assign(:client_id, client_id)
|
||||
|> assign(:bounds, bounds)
|
||||
|> assign(:subscribed, true)
|
||||
|
||||
Logger.info("Mobile client #{client_id} subscribed to bounds: #{inspect(bounds)}")
|
||||
Logger.info("Mobile client #{client_id} subscribed to bounds: #{inspect(bounds)}")
|
||||
|
||||
# Load and send historical packets
|
||||
socket = load_historical_packets(socket, bounds, payload)
|
||||
# Load and send historical packets
|
||||
socket = load_historical_packets(socket, bounds, payload)
|
||||
|
||||
{:reply, {:ok, %{bounds: bounds, message: "Subscribed to packet stream"}}, socket}
|
||||
{:reply, {:ok, %{bounds: bounds, message: "Subscribed to packet stream"}}, socket}
|
||||
|
||||
{:error, reason} ->
|
||||
{:reply, {:error, %{message: reason}}, socket}
|
||||
{:error, reason} ->
|
||||
{:reply, {:error, %{message: reason}}, socket}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -160,37 +167,41 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
def handle_in("search_callsign", %{"query" => query} = payload, socket) do
|
||||
Logger.info("Mobile websocket received search_callsign: #{inspect(payload)}")
|
||||
|
||||
# Trim whitespace and control characters from query
|
||||
query = String.trim(query)
|
||||
with :ok <- check_rate_limit(socket, "search_callsign") do
|
||||
# Trim whitespace and control characters from query
|
||||
query = String.trim(query)
|
||||
|
||||
limit = payload |> Map.get("limit", 50) |> ensure_integer(50)
|
||||
limit = min(limit, 500)
|
||||
limit = payload |> Map.get("limit", 50) |> ensure_integer(50)
|
||||
limit = min(limit, 500)
|
||||
|
||||
results = search_callsign(query, limit)
|
||||
results = search_callsign(query, limit)
|
||||
|
||||
{:reply, {:ok, %{results: results, count: length(results)}}, socket}
|
||||
{:reply, {:ok, %{results: results, count: length(results)}}, socket}
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_in("subscribe_callsign", %{"callsign" => callsign} = payload, socket) do
|
||||
Logger.info("Mobile websocket received subscribe_callsign: #{inspect(payload)}")
|
||||
|
||||
hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24)
|
||||
# Max 1 week
|
||||
hours_back = min(hours_back, 168)
|
||||
with :ok <- check_rate_limit(socket, "subscribe_callsign") do
|
||||
hours_back = payload |> Map.get("hours_back", 24) |> ensure_integer(24)
|
||||
# Max 1 week
|
||||
hours_back = min(hours_back, 168)
|
||||
|
||||
# Normalize callsign - trim whitespace and convert to uppercase
|
||||
callsign = callsign |> String.trim() |> String.upcase()
|
||||
# Normalize callsign - trim whitespace and convert to uppercase
|
||||
callsign = callsign |> String.trim() |> String.upcase()
|
||||
|
||||
# Load historical packets for this callsign
|
||||
socket = load_callsign_history(socket, callsign, hours_back)
|
||||
# Load historical packets for this callsign
|
||||
socket = load_callsign_history(socket, callsign, hours_back)
|
||||
|
||||
socket = ensure_callsign_subscription(socket)
|
||||
socket = ensure_callsign_subscription(socket)
|
||||
|
||||
# Store tracked callsign in socket
|
||||
socket = assign(socket, :tracked_callsign, callsign)
|
||||
# Store tracked callsign in socket
|
||||
socket = assign(socket, :tracked_callsign, callsign)
|
||||
|
||||
{:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket}
|
||||
{:reply, {:ok, %{callsign: callsign, message: "Subscribed to callsign updates"}}, socket}
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
|
|
@ -249,6 +260,40 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
{:noreply, socket}
|
||||
end
|
||||
|
||||
# Per-socket limiter for the expensive client-initiated handlers
|
||||
# (subscribe_bounds, subscribe_callsign, search_callsign). Returns :ok on
|
||||
# allow or a {:reply, ..., socket} tuple that propagates straight out of the
|
||||
# with block to the channel reply path on deny.
|
||||
defp check_rate_limit(socket, op) do
|
||||
key = "mobile_channel:#{op}:#{:erlang.phash2(self())}"
|
||||
|
||||
case Aprsme.RateLimiter.hit(key, @expensive_op_scale, @expensive_op_limit) do
|
||||
{:allow, _count} ->
|
||||
:ok
|
||||
|
||||
{:deny, retry_after} ->
|
||||
Logger.warning("Mobile channel rate-limited op=#{op} client=#{inspect(self())}")
|
||||
|
||||
{:reply, {:error, %{message: "Rate limit exceeded; retry later", retry_after_ms: retry_after}}, socket}
|
||||
end
|
||||
end
|
||||
|
||||
# Push packets to the client in moderately sized batches under the new
|
||||
# `packets` event. Keeps the per-packet `packet` event format intact for
|
||||
# streaming deliveries; bulk historical loads move through this path so
|
||||
# clients don't get hit with 5000 individual WebSocket frames.
|
||||
@packets_batch_size 100
|
||||
defp push_packets_batch(_socket, []), do: :ok
|
||||
|
||||
defp push_packets_batch(socket, packets) do
|
||||
packets
|
||||
|> Stream.map(&build_mobile_packet/1)
|
||||
|> Stream.chunk_every(@packets_batch_size)
|
||||
|> Enum.each(fn chunk ->
|
||||
push(socket, "packets", %{packets: chunk, count: length(chunk)})
|
||||
end)
|
||||
end
|
||||
|
||||
# Clean up on terminate
|
||||
@impl true
|
||||
def terminate(_reason, socket) do
|
||||
|
|
@ -411,14 +456,10 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
|
||||
Logger.info("Loaded #{length(packets)} historical packets for mobile client")
|
||||
|
||||
# Send historical packets to client
|
||||
if packets != [] do
|
||||
# Convert packets to mobile format and send them
|
||||
Enum.each(packets, fn packet ->
|
||||
packet_data = build_mobile_packet(packet)
|
||||
push(socket, "packet", packet_data)
|
||||
end)
|
||||
end
|
||||
# Send historical packets as batched "packets" events rather than one
|
||||
# "packet" frame per row — previously a single subscription could queue
|
||||
# thousands of WebSocket frames.
|
||||
push_packets_batch(socket, packets)
|
||||
|
||||
socket
|
||||
end
|
||||
|
|
@ -508,13 +549,8 @@ defmodule AprsmeWeb.MobileChannel do
|
|||
|
||||
Logger.info("Loaded #{length(packets)} historical packets for callsign #{callsign}")
|
||||
|
||||
# Send historical packets to client
|
||||
if packets != [] do
|
||||
Enum.each(packets, fn packet ->
|
||||
packet_data = build_mobile_packet(packet)
|
||||
push(socket, "packet", packet_data)
|
||||
end)
|
||||
end
|
||||
# Send batched "packets" events — see push_packets_batch/2.
|
||||
push_packets_batch(socket, packets)
|
||||
|
||||
socket
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,36 +1,63 @@
|
|||
defmodule AprsmeWeb.MobileUserSocket do
|
||||
@moduledoc """
|
||||
Socket for mobile applications (iOS/Android) to receive real-time APRS packets.
|
||||
|
||||
Connections are anonymous but rate-limited per peer IP to prevent a single
|
||||
client from opening unbounded connections and exhausting server resources.
|
||||
"""
|
||||
use Phoenix.Socket
|
||||
|
||||
require Logger
|
||||
|
||||
# Channels
|
||||
channel "mobile:packets", AprsmeWeb.MobileChannel
|
||||
|
||||
# Socket params are passed from the client and can be used to verify and authenticate a user.
|
||||
# After verification, you can put default assigns into the socket that will be set for all channels.
|
||||
@impl true
|
||||
def connect(_params, socket, _connect_info) do
|
||||
# For now, allow anonymous connections
|
||||
# In the future, you can add authentication here:
|
||||
# case verify_token(params["token"]) do
|
||||
# {:ok, user_id} -> {:ok, assign(socket, :user_id, user_id)}
|
||||
# {:error, _} -> :error
|
||||
# end
|
||||
# 30 new connections per IP per minute — well above normal reconnect churn
|
||||
# but tight enough to stop a script from floodting the socket layer.
|
||||
@connect_limit 30
|
||||
@connect_scale 60_000
|
||||
|
||||
{:ok, socket}
|
||||
@impl true
|
||||
def connect(_params, socket, connect_info) do
|
||||
ip = peer_ip(connect_info)
|
||||
|
||||
if test_env?() or allow_connect?(ip) do
|
||||
{:ok, assign(socket, :peer_ip, ip)}
|
||||
else
|
||||
Logger.warning("Mobile socket connect rate-limited for ip=#{ip}")
|
||||
:error
|
||||
end
|
||||
end
|
||||
|
||||
# Socket id's are topics that allow you to identify all sockets for a given user:
|
||||
#
|
||||
# def id(socket), do: "mobile_user_socket:#{socket.assigns.user_id}"
|
||||
#
|
||||
# Would allow you to broadcast a "disconnect" event and terminate
|
||||
# all active sockets and channels for a given user:
|
||||
#
|
||||
# Elixir.AprsmeWeb.Endpoint.broadcast("mobile_user_socket:#{user.id}", "disconnect", %{})
|
||||
#
|
||||
# Returning `nil` makes this socket anonymous.
|
||||
defp allow_connect?(ip) do
|
||||
case Aprsme.RateLimiter.hit("mobile_socket_connect:#{ip}", @connect_scale, @connect_limit) do
|
||||
{:allow, _count} -> true
|
||||
{:deny, _retry_after} -> false
|
||||
end
|
||||
end
|
||||
|
||||
defp test_env?, do: Application.get_env(:aprsme, :env) == :test
|
||||
|
||||
@impl true
|
||||
def id(_socket), do: nil
|
||||
|
||||
defp peer_ip(%{x_headers: headers}) when is_list(headers) do
|
||||
headers
|
||||
|> Enum.find_value(fn
|
||||
{"cf-connecting-ip", v} -> v
|
||||
{"x-forwarded-for", v} -> v |> String.split(",") |> List.first() |> String.trim()
|
||||
{"x-real-ip", v} -> v
|
||||
_ -> nil
|
||||
end)
|
||||
|> case do
|
||||
nil -> "unknown"
|
||||
ip -> ip
|
||||
end
|
||||
end
|
||||
|
||||
defp peer_ip(%{peer_data: %{address: address}}) when not is_nil(address) do
|
||||
address |> :inet.ntoa() |> to_string()
|
||||
end
|
||||
|
||||
defp peer_ip(_), do: "unknown"
|
||||
end
|
||||
|
|
|
|||
|
|
@ -52,8 +52,12 @@ defmodule AprsmeWeb.MobileChannelTest do
|
|||
end
|
||||
|
||||
describe "join" do
|
||||
test "successful join returns welcome message", %{socket: socket} do
|
||||
assert socket.assigns == %{}
|
||||
test "successful join stamps the peer ip and has no other assigns", %{socket: socket} do
|
||||
# Socket test client has no real peer_data, so the IP helper falls back
|
||||
# to "unknown"; the important part is that the assign is present and
|
||||
# no extra per-subscription state has been set yet.
|
||||
assert socket.assigns.peer_ip == "unknown"
|
||||
assert Map.delete(socket.assigns, :peer_ip) == %{}
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue