GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Send configured request headers with a GTFS feed's requests

7753009 · Gabriel Morell · 2026-09-07 04:30

23 files +286 -34
Message 
{commit_body(@commit)}

Files changed

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified apps/room_gtfs/lib/room_gtfs/feed_cache.ex
+42 −28
@@ -73,23 +73,37 @@ defmodule RoomGtfs.FeedCache do
73 73 @doc """
74 74 The parsed feed at `url`, fetching it only if nobody has within `max_age`.
75 75
76 − Returns what `RoomGtfs.Worker.RT.fetch_rt_url/1` returns, so it drops into
76 + Returns what `RoomGtfs.Worker.RT.fetch_rt_url/2` returns, so it drops into
77 77 call sites unchanged: `{:ok, %TransitRealtime.FeedMessage{}}` or
78 78 `{:error, reason}`. Errors are not cached -- a feed that failed should be
79 79 retried on the next poll, not remembered as broken for 25 seconds.
80 +
81 + `headers` are sent with the fetch; `scope` is who is asking, and entries are
82 + keyed by both the URL and it. A feed can require an API key, and two people
83 + reading the same feed hold different ones -- so the sharing this exists for
84 + stops at the owner's edge. Within one owner nothing changes: several sources
85 + pointing at one URL still fetch it once between them.
86 +
87 + Scoping by owner rather than by the headers themselves is deliberate. It
88 + keeps one entry per URL per person however their credentials are written,
89 + and it keeps secrets out of an ETS key that `stats/0` and any observer can
90 + read back.
80 91 """
81 − def get(url, max_age \\ @default_max_age) when is_binary(url) do
82 − case fresh_entry(url, max_age) do
92 + def get(url, max_age \\ @default_max_age, headers \\ [], scope \\ nil)
93 + when is_binary(url) do
94 + key = {url, scope}
95 +
96 + case fresh_entry(key, max_age) do
83 97 {:ok, _feed} = hit ->
84 98 bump(@hits)
85 99 hit
86 100
87 101 :stale ->
88 − GenServer.call(__MODULE__, {:fetch, url, max_age}, @call_timeout)
102 + GenServer.call(__MODULE__, {:fetch, key, headers, max_age}, @call_timeout)
89 103 end
90 104 catch
91 105 :exit, _reason ->
92 − RoomGtfs.Worker.RT.fetch_rt_url(url)
106 + RoomGtfs.Worker.RT.fetch_rt_url(url, headers)
93 107 end
94 108
95 109 @doc """
@@ -119,41 +133,41 @@ defmodule RoomGtfs.FeedCache do
119 133 end
120 134
121 135 @impl true
122 − def handle_call({:fetch, url, max_age}, from, state) do
136 + def handle_call({:fetch, key, headers, max_age}, from, state) do
123 137 # Re-checked: the caller missed, then waited on this mailbox, and another
124 − # fetch of the same URL may have landed in between.
125 − case fresh_entry(url, max_age) do
138 + # fetch of the same key may have landed in between.
139 + case fresh_entry(key, max_age) do
126 140 {:ok, _feed} = hit ->
127 141 bump(@hits)
128 142 {:reply, hit, state}
129 143
130 144 :stale ->
131 145 case state.inflight do
132 − %{^url => waiting} ->
133 − {:noreply, put_in(state.inflight[url], [from | waiting])}
146 + %{^key => waiting} ->
147 + {:noreply, put_in(state.inflight[key], [from | waiting])}
134 148
135 149 _ ->
136 − {_pid, ref} = spawn_fetch(url)
150 + {_pid, ref} = spawn_fetch(key, headers)
137 151
138 152 {:noreply,
139 153 state
140 − |> put_in([:inflight, url], [from])
141 − |> put_in([:refs, ref], url)}
154 + |> put_in([:inflight, key], [from])
155 + |> put_in([:refs, ref], key)}
142 156 end
143 157 end
144 158 end
145 159
146 160 @impl true
147 − def handle_info({:fetched, url, result}, state) do
148 − {waiting, inflight} = Map.pop(state.inflight, url, [])
161 + def handle_info({:fetched, key, result}, state) do
162 + {waiting, inflight} = Map.pop(state.inflight, key, [])
149 163
150 164 # Dropped before the fetcher's :DOWN arrives, so the normal exit that
151 165 # follows this message is not mistaken for a crash.
152 − {refs, _} = pop_ref(state.refs, url)
166 + {refs, _} = pop_ref(state.refs, key)
153 167
154 168 case result do
155 169 {:ok, _feed} ->
156 − :ets.insert(@table, {url, System.monotonic_time(:millisecond), result})
170 + :ets.insert(@table, {key, System.monotonic_time(:millisecond), result})
157 171 bump(@fetches)
158 172
159 173 {:error, _reason} ->
@@ -182,9 +196,9 @@ defmodule RoomGtfs.FeedCache do
182 196 {nil, _refs} ->
183 197 {:noreply, state}
184 198
185 − {url, refs} ->
199 + {{url, _scope} = key, refs} ->
186 200 Logger.warning("gtfs-rt fetch of #{url} died: #{inspect(reason)}")
187 − {waiting, inflight} = Map.pop(state.inflight, url, [])
201 + {waiting, inflight} = Map.pop(state.inflight, key, [])
188 202 Enum.each(waiting, &GenServer.reply(&1, {:error, {:fetch_crashed, reason}}))
189 203 {:noreply, %{state | inflight: inflight, refs: refs}}
190 204 end
@@ -193,32 +207,32 @@ defmodule RoomGtfs.FeedCache do
193 207 @impl true
194 208 def handle_info(_msg, state), do: {:noreply, state}
195 209
196 − defp pop_ref(refs, url) do
197 − case Enum.find(refs, fn {_ref, u} -> u == url end) do
210 + defp pop_ref(refs, key) do
211 + case Enum.find(refs, fn {_ref, k} -> k == key end) do
198 212 nil -> {refs, nil}
199 213 {ref, _} -> {Map.delete(refs, ref), ref}
200 214 end
201 215 end
202 216
203 − defp spawn_fetch(url) do
217 + defp spawn_fetch({url, _scope} = key, headers) do
204 218 parent = self()
205 219
206 220 spawn_monitor(fn ->
207 221 result =
208 222 try do
209 − RoomGtfs.Worker.RT.fetch_rt_url(url)
223 + RoomGtfs.Worker.RT.fetch_rt_url(url, headers)
210 224 rescue
211 225 e -> {:error, e}
212 226 catch
213 227 kind, reason -> {:error, {kind, reason}}
214 228 end
215 229
216 − send(parent, {:fetched, url, result})
230 + send(parent, {:fetched, key, result})
217 231 end)
218 232 end
219 233
220 − defp fresh_entry(url, max_age) do
221 − with [{^url, at, result}] <- safe_lookup(url),
234 + defp fresh_entry(key, max_age) do
235 + with [{^key, at, result}] <- safe_lookup(key),
222 236 true <- System.monotonic_time(:millisecond) - at <= max_age do
223 237 result
224 238 else
@@ -226,8 +240,8 @@ defmodule RoomGtfs.FeedCache do
226 240 end
227 241 end
228 242
229 − defp safe_lookup(url) do
230 − :ets.lookup(@table, url)
243 + defp safe_lookup(key) do
244 + :ets.lookup(@table, key)
231 245 rescue
232 246 ArgumentError -> []
233 247 end
modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+22 −4
@@ -1151,8 +1151,8 @@ defmodule RoomGtfs.Worker.RT do
1151 1151 end
1152 1152
1153 1153
1154 − def fetch_rt_url(url) do
1155 − case HTTPoison.get(url, [], @rt_http) do
1154 + def fetch_rt_url(url, headers \\ []) do
1155 + case HTTPoison.get(url, headers, @rt_http) do
1156 1156 {:ok, %{status_code: 200} = result} ->
1157 1157 decode_rt(url, result)
1158 1158
@@ -1380,7 +1380,23 @@ defmodule RoomGtfs.Worker.RT do
1380 1380
1381 1381 defp do_fetch_group(state, url, kinds, period) do
1382 1382 started = System.monotonic_time(:millisecond)
1383 − result = FeedCache.get(url, max(period - :timer.seconds(5), :timer.seconds(1)))
1383 +
1384 + headers =
1385 + RoomSanctum.Configuration.Configs.GTFS.request_headers(
1386 + state.inst && state.inst.config
1387 + )
1388 +
1389 + # Scoped to the owner. Several of one person's sources pointing at one URL
1390 + # still fetch it once between them, which is what this cache is for; two
1391 + # people's do not, because they may hold different credentials for it and
1392 + # a hit fetched with somebody else's key is a bad kind of correct.
1393 + result =
1394 + FeedCache.get(
1395 + url,
1396 + max(period - :timer.seconds(5), :timer.seconds(1)),
1397 + headers,
1398 + state.inst && state.inst.user_id
1399 + )
1384 1400 took = System.monotonic_time(:millisecond) - started
1385 1401
1386 1402 case result do
@@ -2330,7 +2346,9 @@ end
2330 2346 Logger.info("GTFS::#{id} updating static info")
2331 2347 bcast(id, :downloading, 1, 11)
2332 2348
2333 − case HTTPoison.get(cfg.config.url, [], @static_http) do
2349 + headers = RoomSanctum.Configuration.Configs.GTFS.request_headers(cfg.config)
2350 +
2351 + case HTTPoison.get(cfg.config.url, headers, @static_http) do
2334 2352 {:ok, %{status_code: 200} = result} ->
2335 2353 bcast(id, :extracting, 2, 11)
2336 2354 Logger.info("GTFS::#{id} downloaded #{byte_size(result.body)} bytes")
modified apps/room_sanctum/lib/room_sanctum/configuration/configs/gtfs.ex
+101 −1
@@ -49,6 +49,19 @@ defmodule RoomSanctum.Configuration.Configs.GTFS do
49 49 # looser than an exact one and can match a trip it should not; a feed that
50 50 # needs it should say so rather than have it guessed from a bad day's data.
51 51 field :rt_trip_id_suffix, :boolean, default: false
52 +
53 + # Headers to send with every request for this feed: static zip, realtime,
54 + # and the linked datasets a discovery file points at.
55 + #
56 + # Arbitrary because the schemes are. Some agencies want `apikey`, some
57 + # `x-api-key`, 511 wants it in the query string and needs none of this,
58 + # and TMB wants `Authorization: Bearer ...`. Naming a scheme here would
59 + # mean a new field for every operator with an opinion.
60 + field :headers, :map, default: %{}
61 +
62 + # What the form edits: one `Name: value` per line. The stored shape is the
63 + # map; this is the shape a person can type and read back.
64 + field :headers_raw, :string, virtual: true
52 65 end
53 66
54 67 def changeset(source, params) do
@@ -56,10 +69,97 @@ defmodule RoomSanctum.Configuration.Configs.GTFS do
56 69 |> cast(
57 70 params,
58 71 ~w(url url_rt_sa url_rt_tu url_rt_vp url_rt_shared tz rt_agency
59 − rt_trip_id_suffix rt_period_tu rt_period_vp rt_period_sa)a
72 + rt_trip_id_suffix rt_period_tu rt_period_vp rt_period_sa headers)a
60 73 )
74 + # Its own cast, with "" kept as a value rather than treated as nothing.
75 + # Ecto's default empty_values turns an emptied textarea into no change at
76 + # all, so clearing the box left the old headers in place -- which means an
77 + # API key could be added through this form and never taken away again.
78 + |> cast(params, [:headers_raw], empty_values: [])
61 79 |> validate_required([:url, :tz])
62 80 |> validate_periods()
81 + |> put_headers()
82 + end
83 +
84 + @doc """
85 + The configured headers in the shape HTTPoison wants.
86 +
87 + `[]` for a source that sets none, which is nearly all of them -- and which
88 + is also what every caller passed before this existed, so an unconfigured
89 + feed makes exactly the request it always did.
90 + """
91 + def request_headers(%{headers: headers}) when is_map(headers) do
92 + Enum.map(headers, fn {name, value} -> {to_string(name), to_string(value)} end)
93 + end
94 +
95 + def request_headers(_config), do: []
96 +
97 + @doc """
98 + The stored headers as the text the form edits: one `Name: value` per line.
99 +
100 + Sorted, so editing a source does not reshuffle the box every time it is
101 + opened -- a map has no order and a textarea that reorders itself looks like
102 + it changed something.
103 + """
104 + def headers_text(%{headers: headers}) when is_map(headers) and map_size(headers) > 0 do
105 + headers
106 + |> Enum.sort_by(fn {name, _value} -> name end)
107 + |> Enum.map_join("\n", fn {name, value} -> "#{name}: #{value}" end)
108 + end
109 +
110 + def headers_text(_config), do: ""
111 +
112 + @doc """
113 + Parse the textarea into a map, or say which line is wrong.
114 +
115 + Returns `{:ok, map}` or `{:error, line}`. Public because the form and the
116 + changeset both want it and a second copy of "what a header line looks like"
117 + is the kind of thing that drifts.
118 + """
119 + def parse_headers(text) when is_binary(text) do
120 + text
121 + |> String.split(~r/\r?\n/)
122 + |> Enum.map(&String.trim/1)
123 + |> Enum.reject(&(&1 == ""))
124 + |> Enum.reduce_while({:ok, %{}}, fn line, {:ok, acc} ->
125 + case String.split(line, ":", parts: 2) do
126 + [name, value] ->
127 + name = String.trim(name)
128 + value = String.trim(value)
129 +
130 + # A name with a space in it is not a header name, and is nearly
131 + # always a colon typed in the value of the line above.
132 + if name != "" and not String.contains?(name, " ") do
133 + {:cont, {:ok, Map.put(acc, name, value)}}
134 + else
135 + {:halt, {:error, line}}
136 + end
137 +
138 + _ ->
139 + {:halt, {:error, line}}
140 + end
141 + end)
142 + end
143 +
144 + def parse_headers(_text), do: {:ok, %{}}
145 +
146 + # Only when the form sent the text field. A changeset that does not mention
147 + # headers_raw -- anything not coming from that form -- leaves the stored
148 + # headers exactly as they were.
149 + defp put_headers(changeset) do
150 + case fetch_change(changeset, :headers_raw) do
151 + :error ->
152 + changeset
153 +
154 + {:ok, text} ->
155 + case parse_headers(text) do
156 + {:ok, headers} ->
157 + put_change(changeset, :headers, headers)
158 +
159 + {:error, line} ->
160 + add_error(changeset, :headers_raw, "expected \"Name: value\", got: #{line}")
161 + end
162 + end
63 163 end
64 164
65 165 # Below about ten seconds the poller cannot keep up and the feed will not have
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.html.heex
+19 −0
@@ -125,6 +125,25 @@
125 125 Roughly <span class="font-bold"><%= @rt_requests %></span> requests an hour.
126 126 511.org allows 60 on a default token.
127 127 </div>
128 + <div class="form-control">
129 + <.input
130 + field={gtfs_form[:headers_raw]}
131 + type="textarea"
132 + label="Request headers"
133 + rows="3"
134 + value={gtfs_form[:headers_raw].value ||
135 + RoomSanctum.Configuration.Configs.GTFS.headers_text(gtfs_form.data)}
136 + placeholder={"Authorization: Bearer ...\nx-api-key: ..."}
137 + />
138 + <label class="label">
139 + <span class="label-text-alt opacity-70">
140 + One <code>Name: value</code> per line, sent with every request for this
141 + feed &mdash; the static zip and the realtime feeds alike. For agencies
142 + that gate downloads behind a key. Leave empty for the ones that do not,
143 + which is most of them.
144 + </span>
145 + </label>
146 + </div>
128 147 <div class="form-control">
129 148 <.input field={gtfs_form[:rt_trip_id_suffix]} type="checkbox"
130 149 label="Realtime sends partial trip ids"/>
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/show.ex
+4 −1
@@ -177,7 +177,10 @@ defmodule RoomSanctumWeb.SourceLive.Show do
177 177
178 178 @impl true
179 179 def handle_params(%{"id" => id}, _, socket) do
180 − source = Configuration.get_source!(id) |> IO.inspect
180 + # Not inspected. This struct now carries a source's request headers, and an
181 + # API key printed to the logs on every page load is a secret in a place
182 + # nobody thinks to look for one.
183 + source = Configuration.get_source!(id)
181 184
182 185 # Subscribe to relevant PubSub channels
183 186 case source.type do
added apps/room_sanctum/test/room_sanctum/gtfs_headers_test.exs
+98 −0
@@ -0,0 +1,98 @@
1 +defmodule RoomSanctum.GtfsHeadersTest do
2 + use ExUnit.Case, async: true
3 +
4 + alias Ecto.Changeset
5 + alias RoomSanctum.Configuration.Configs.GTFS
6 +
7 + defp change(attrs) do
8 + GTFS.changeset(%GTFS{}, Map.merge(%{"url" => "http://x/gtfs.zip", "tz" => "UTC"}, attrs))
9 + end
10 +
11 + describe "parsing what a person typed" do
12 + test "one header per line" do
13 + assert {:ok, %{"Authorization" => "Bearer abc", "x-api-key" => "k"}} =
14 + GTFS.parse_headers("Authorization: Bearer abc\nx-api-key: k")
15 + end
16 +
17 + test "a value may contain colons" do
18 + # Split on the first colon only: a Bearer token or a URL in a header
19 + # value is not two headers.
20 + assert {:ok, %{"Authorization" => "Bearer a:b:c"}} =
21 + GTFS.parse_headers("Authorization: Bearer a:b:c")
22 + end
23 +
24 + test "blank lines and stray whitespace are not headers" do
25 + assert {:ok, %{"apikey" => "spaced"}} = GTFS.parse_headers("\n apikey: spaced \n\n")
26 + end
27 +
28 + test "a line with no colon is an error naming the line" do
29 + assert {:error, "no colon here"} = GTFS.parse_headers("no colon here")
30 + end
31 +
32 + test "a name with a space in it is rejected" do
33 + # Nearly always a colon typed into the previous line's value.
34 + assert {:error, _} = GTFS.parse_headers("bad name: x")
35 + end
36 +
37 + test "nothing typed is no headers, not an error" do
38 + assert {:ok, %{}} = GTFS.parse_headers("")
39 + end
40 + end
41 +
42 + describe "the changeset" do
43 + test "valid text becomes the stored map" do
44 + assert %{"Authorization" => "Bearer zzz"} =
45 + change(%{"headers_raw" => "Authorization: Bearer zzz"})
46 + |> Changeset.get_field(:headers)
47 + end
48 +
49 + test "a bad line fails the changeset rather than storing half of it" do
50 + changeset = change(%{"headers_raw" => "oops"})
51 +
52 + refute changeset.valid?
53 + assert changeset.errors[:headers_raw]
54 + end
55 +
56 + test "a form that does not mention headers leaves them alone" do
57 + # Anything editing a source by another route must not wipe its key.
58 + existing = %GTFS{headers: %{"keep" => "me"}}
59 +
60 + assert %{"keep" => "me"} =
61 + GTFS.changeset(existing, %{"url" => "http://x", "tz" => "UTC"})
62 + |> Changeset.get_field(:headers)
63 + end
64 +
65 + test "clearing the box clears the headers" do
66 + existing = %GTFS{headers: %{"gone" => "soon"}}
67 +
68 + assert %{} ==
69 + GTFS.changeset(existing, %{"url" => "http://x", "tz" => "UTC", "headers_raw" => ""})
70 + |> Changeset.get_field(:headers)
71 + end
72 + end
73 +
74 + describe "what the worker sends" do
75 + test "the stored map becomes a header list" do
76 + assert [{"x-api-key", "k"}] = GTFS.request_headers(%{headers: %{"x-api-key" => "k"}})
77 + end
78 +
79 + test "a source with none makes the request it always made" do
80 + assert [] == GTFS.request_headers(%{headers: %{}})
81 + assert [] == GTFS.request_headers(%{})
82 + assert [] == GTFS.request_headers(nil)
83 + end
84 + end
85 +
86 + describe "round tripping" do
87 + test "stored headers render back as the text that made them" do
88 + text = "Authorization: Bearer abc\nx-api-key: k"
89 + {:ok, parsed} = GTFS.parse_headers(text)
90 +
91 + assert GTFS.headers_text(%{headers: parsed}) == text
92 + end
93 +
94 + test "the text is ordered, so opening a form twice shows the same thing" do
95 + assert GTFS.headers_text(%{headers: %{"b" => "2", "a" => "1"}}) == "a: 1\nb: 2"
96 + end
97 + end
98 +end

Parents: 4c3c2fc