defmodule GitGud.Federation.HttpSignatures do
@moduledoc """
HTTP Signatures support per draft-cavage-http-signatures-12 — the
flavour that ActivityPub clients (Mastodon, Pleroma, Forgejo, …) all
speak.
Currently supports:
* `rsa-sha256` (the only ubiquitous algorithm)
* Signing string composition over `(request-target)`, `host`,
`date`, `digest`, plus arbitrary additional headers
* Digest header: `SHA-256=<base64>` of the request body
Caller flow for inbound:
with {:ok, params} <- parse_header(sig_header),
{:ok, actor} <- resolve_key(params["keyId"]),
:ok <- verify_digest(headers, body),
:ok <- verify_clock_skew(headers),
:ok <- verify(actor.public_key_pem,
signing_string, params["signature"]) do
:ok
end
Caller flow for outbound:
headers = sign_request("post", path, host, body, actor_private_pem,
actor.public_key_id)
# → list of {header_name, value} including Date, Digest, Signature
"""
@max_clock_skew_seconds 60 * 5
# ── parsing ──────────────────────────────────────────────────────────
@doc """
Parse a `Signature: keyId="...",algorithm="...",headers="...",signature="..."`
header into a map of strings.
"""
def parse_header(header) when is_binary(header) do
pairs =
header
|> String.split(",", trim: true)
|> Enum.map(&parse_pair/1)
|> Enum.reject(&is_nil/1)
|> Map.new()
if Map.has_key?(pairs, "keyId") and Map.has_key?(pairs, "signature") do
{:ok, pairs}
else
{:error, :malformed_signature}
end
end
defp parse_pair(s) do
case String.split(String.trim(s), "=", parts: 2) do
[k, v] -> {String.trim(k), unquote_str(v)}
_ -> nil
end
end
defp unquote_str(<<"\"", rest::binary>>) do
case String.split(rest, "\"", parts: 2) do
[val, _] -> val
[val] -> val
end
end
defp unquote_str(s), do: s
# ── signing-string composition ───────────────────────────────────────
@doc """
Build the canonical signing string from a list of headers and the
pseudo-header `(request-target)`.
`headers_param` is the space-separated list from the Signature
header's `headers=` parameter, e.g.
`"(request-target) host date digest"`. `header_lookup` is a function
`fn name -> value | nil` that returns the request header value.
`request_target` is `"<method> <path>"`, all lowercase.
"""
def signing_string(headers_param, header_lookup, request_target) do
headers_param
|> String.split(" ", trim: true)
|> Enum.map(fn
"(request-target)" -> "(request-target): " <> request_target
name -> "#{String.downcase(name)}: #{header_lookup.(name) || ""}"
end)
|> Enum.join("\n")
end
# ── verification ─────────────────────────────────────────────────────
@doc "Verify an RSA-SHA256 signature against a PEM-encoded public key."
def verify(public_key_pem, signing_string, signature_b64)
when is_binary(public_key_pem) and is_binary(signing_string) and is_binary(signature_b64) do
with {:ok, signature} <- Base.decode64(signature_b64),
{:ok, key} <- decode_public_key(public_key_pem) do
if :public_key.verify(signing_string, :sha256, signature, key) do
:ok
else
{:error, :bad_signature}
end
else
:error -> {:error, :bad_signature_b64}
{:error, _} = err -> err
end
end
defp decode_public_key(pem) do
case :public_key.pem_decode(pem) do
[entry | _] ->
case :public_key.pem_entry_decode(entry) do
{:RSAPublicKey, _, _} = k -> {:ok, k}
other -> {:ok, other}
end
[] ->
{:error, :no_key}
end
end
@doc """
Verify that the request body matches the value in the `Digest`
header. `digest_header` looks like `SHA-256=<base64>`.
"""
def verify_digest(nil, _body), do: {:error, :missing_digest}
def verify_digest(digest_header, body) when is_binary(body) do
case String.split(digest_header, "=", parts: 2) do
["SHA-256", b64] ->
case Base.decode64(b64) do
{:ok, expected} ->
actual = :crypto.hash(:sha256, body)
if expected == actual, do: :ok, else: {:error, :digest_mismatch}
:error ->
{:error, :bad_digest_b64}
end
_ ->
{:error, :unsupported_digest}
end
end
@doc """
Reject signatures whose `Date` header is too far from now. Mitigates
replay attacks. `now` defaults to system time.
"""
def verify_clock_skew(date_header, now \\ DateTime.utc_now()) do
case parse_http_date(date_header) do
{:ok, dt} ->
if abs(DateTime.diff(dt, now)) <= @max_clock_skew_seconds,
do: :ok,
else: {:error, :clock_skew}
_ ->
{:error, :bad_date}
end
end
defp parse_http_date(nil), do: :error
defp parse_http_date(s) when is_binary(s), do: parse_imf_fixdate(s)
defp parse_imf_fixdate(s) do
# "Sun, 06 Nov 1994 08:49:37 GMT" — what HTTP actually sends.
case Regex.run(
~r/^[A-Za-z]+,\s+(\d{1,2})\s+([A-Za-z]+)\s+(\d{4})\s+(\d{2}):(\d{2}):(\d{2})\s+GMT$/,
s
) do
[_, day, mon, year, h, m, sec] ->
with month <- month_to_int(mon),
{:ok, dt} <-
DateTime.new(
Date.new!(String.to_integer(year), month, String.to_integer(day)),
Time.new!(String.to_integer(h), String.to_integer(m), String.to_integer(sec)),
"Etc/UTC"
) do
{:ok, dt}
else
_ -> :error
end
_ ->
:error
end
end
defp month_to_int("Jan"), do: 1
defp month_to_int("Feb"), do: 2
defp month_to_int("Mar"), do: 3
defp month_to_int("Apr"), do: 4
defp month_to_int("May"), do: 5
defp month_to_int("Jun"), do: 6
defp month_to_int("Jul"), do: 7
defp month_to_int("Aug"), do: 8
defp month_to_int("Sep"), do: 9
defp month_to_int("Oct"), do: 10
defp month_to_int("Nov"), do: 11
defp month_to_int("Dec"), do: 12
defp month_to_int(_), do: 0
# ── signing (outbound) ───────────────────────────────────────────────
@doc """
Build the headers a signed AP POST needs. Returns a keyword list of
`{name, value}` to merge into the outgoing request.
Required arguments:
* `method` — `"post"` or `"get"`, lowercase
* `path` — request-target path (with query if any)
* `host` — target host (with port if non-default)
* `body` — the bytes that will be sent (use `""` for GET)
* `private_key_pem` — the sending actor's private key
* `public_key_id` — the sender's publicKey URL (used as keyId)
"""
def sign_request(method, path, host, body, private_key_pem, public_key_id) do
now = http_date(DateTime.utc_now())
digest = "SHA-256=" <> Base.encode64(:crypto.hash(:sha256, body))
header_lookup = fn
"host" -> host
"date" -> now
"digest" -> digest
_ -> nil
end
headers_list =
case method do
"get" -> "(request-target) host date"
_ -> "(request-target) host date digest"
end
signing_str = signing_string(headers_list, header_lookup, "#{method} #{path}")
signature = sign(private_key_pem, signing_str) |> Base.encode64()
sig_header =
~s(keyId="#{public_key_id}",algorithm="rsa-sha256",headers="#{headers_list}",signature="#{signature}")
base = [
{"host", host},
{"date", now},
{"signature", sig_header}
]
case method do
"get" -> base
_ -> base ++ [{"digest", digest}]
end
end
defp sign(private_key_pem, signing_string) do
[entry] = :public_key.pem_decode(private_key_pem)
key = :public_key.pem_entry_decode(entry)
:public_key.sign(signing_string, :sha256, key)
end
defp http_date(%DateTime{} = dt) do
# IMF-fixdate, e.g. "Sun, 06 Nov 1994 08:49:37 GMT"
Calendar.strftime(dt, "%a, %d %b %Y %H:%M:%S GMT")
end
end
8.3 KiB · text
5af55d9