diff --git a/apps/elektrine/lib/elektrine/application.ex b/apps/elektrine/lib/elektrine/application.ex index 5cfef66..eab066c 100644 --- a/apps/elektrine/lib/elektrine/application.ex +++ b/apps/elektrine/lib/elektrine/application.ex @@ -83,7 +83,20 @@ defmodule Elektrine.Application do Elektrine.SecurityAlerts.Cache, Elektrine.CaddyTLSDomainCache, ElektrineWeb.Presence - ] ++ ModuleProviders.web_children() ++ [ElektrineWeb.Endpoint] + ] ++ + edge_connector_children() ++ + ModuleProviders.web_children() ++ [ElektrineWeb.Endpoint] + else + [] + end + end + + # Edge tunnel connector must run on web-enabled nodes only (KD-5c). + defp edge_connector_children do + connector = ElektrineWeb.Edge.Connector + + if Code.ensure_loaded?(connector) do + [connector] else [] end diff --git a/apps/elektrine/lib/elektrine/developer/api_token.ex b/apps/elektrine/lib/elektrine/developer/api_token.ex index 02d5012..b798ade 100644 --- a/apps/elektrine/lib/elektrine/developer/api_token.ex +++ b/apps/elektrine/lib/elektrine/developer/api_token.ex @@ -53,6 +53,7 @@ defmodule Elektrine.Developer.ApiToken do read:nerve write:nerve read:kairo write:kairo read:dns write:dns + read:edge write:edge read:proofs write:proofs read:static_site write:static_site read:moderation write:moderation @@ -108,6 +109,12 @@ defmodule Elektrine.Developer.ApiToken do description: "Read and write managed DNS zones and records.", scopes: ["read:dns", "write:dns"] }, + %{ + id: "edge_tunnels", + name: "Edge tunnels", + description: "Mint, list, and revoke private origin tunnels and edge cache purge.", + scopes: ["read:edge", "write:edge"] + }, %{ id: "identity_proofs", name: "Identity proofs", @@ -227,6 +234,10 @@ defmodule Elektrine.Developer.ApiToken do {"read:dns", "Read managed DNS zones and records"}, {"write:dns", "Create, update, verify, and delete DNS zones and records"} ], + "Edge" => [ + {"read:edge", "List edge tunnels and related zone edge config"}, + {"write:edge", "Mint/revoke tunnels and purge edge cache"} + ], "Identity" => [ {"read:proofs", "Read identity proofs and personhood score"}, {"write:proofs", "Create, check, and delete identity proofs"} diff --git a/apps/elektrine/mix.exs b/apps/elektrine/mix.exs index bdf8185..d7791ed 100644 --- a/apps/elektrine/mix.exs +++ b/apps/elektrine/mix.exs @@ -67,6 +67,7 @@ defmodule Elektrine.MixProject do Elektrine.Social.PostBoost, Elektrine.Social.PostLike, ElektrineWeb.Endpoint, + ElektrineWeb.Edge.Connector, ElektrineEmailWeb.HarakaWebhookController, ElektrineWeb.Presence ] diff --git a/apps/elektrine/priv/repo/migrations/20260802010000_create_dns_tunnels.exs b/apps/elektrine/priv/repo/migrations/20260802010000_create_dns_tunnels.exs new file mode 100644 index 0000000..c6196ba --- /dev/null +++ b/apps/elektrine/priv/repo/migrations/20260802010000_create_dns_tunnels.exs @@ -0,0 +1,25 @@ +defmodule Elektrine.Repo.Migrations.CreateDnsTunnels do + use Ecto.Migration + + def change do + create table(:dns_tunnels, primary_key: false) do + add :id, :binary_id, primary_key: true + add :zone_id, references(:dns_zones, on_delete: :delete_all), null: false + add :name, :string, null: false + add :status, :string, null: false, default: "inactive" + add :allowed_hosts, {:array, :string}, null: false, default: [] + add :token_hash, :string, null: false + add :token_prefix, :string, null: false + add :created_by, references(:users, on_delete: :nilify_all) + add :last_seen_at, :utc_datetime + add :revoked_at, :utc_datetime + + timestamps(type: :utc_datetime) + end + + create index(:dns_tunnels, [:zone_id]) + create index(:dns_tunnels, [:status]) + create unique_index(:dns_tunnels, [:token_hash]) + create unique_index(:dns_tunnels, [:zone_id, :name], name: :dns_tunnels_zone_id_name_unique) + end +end diff --git a/apps/elektrine/priv/static/openapi/ext-v1.yaml b/apps/elektrine/priv/static/openapi/ext-v1.yaml index 31c69d6..4a50f64 100644 --- a/apps/elektrine/priv/static/openapi/ext-v1.yaml +++ b/apps/elektrine/priv/static/openapi/ext-v1.yaml @@ -730,6 +730,12 @@ paths: parameters: - in: path name: id +/api/ext/v1/dns/zones/{zone_id}/tunnels: + get: + summary: List edge tunnels for a zone (read:edge) + parameters: + - in: path + name: zone_id required: true schema: type: integer @@ -741,6 +747,12 @@ paths: parameters: - in: path name: id +description: Tunnel list (no secrets) + post: + summary: Mint an edge tunnel (write:edge); token shown once + parameters: + - in: path + name: zone_id required: true schema: type: integer @@ -858,6 +870,69 @@ paths: responses: "200": description: Edge rule deleted +required: [name] + properties: + name: + type: string + allowed_hosts: + type: array + items: + type: string + responses: + "201": + description: Tunnel created with one-time etn_ token + /api/ext/v1/dns/zones/{zone_id}/tunnels/{id}: + get: + summary: Show an edge tunnel (read:edge) + parameters: + - in: path + name: zone_id + required: true + schema: + type: integer + - in: path + name: id + required: true + schema: + type: string + format: uuid + responses: + "200": + description: Tunnel details + put: + summary: Update an edge tunnel (write:edge) + parameters: + - in: path + name: zone_id + required: true + schema: + type: integer + - in: path + name: id + required: true + schema: + type: string + format: uuid + responses: + "200": + description: Tunnel updated + delete: + summary: Revoke an edge tunnel (write:edge) + parameters: + - in: path + name: zone_id + required: true + schema: + type: integer + - in: path + name: id + required: true + schema: + type: string + format: uuid + responses: + "200": + description: Tunnel revoked /api/ext/v1/nerve/entries: get: summary: List encrypted nerve entries diff --git a/apps/elektrine_dns/lib/elektrine/dns.ex b/apps/elektrine_dns/lib/elektrine/dns.ex index 08ead31..468d0e5 100644 --- a/apps/elektrine_dns/lib/elektrine/dns.ex +++ b/apps/elektrine_dns/lib/elektrine/dns.ex @@ -15,6 +15,7 @@ defmodule Elektrine.DNS do alias Elektrine.DNS.Packet alias Elektrine.DNS.QueryStat alias Elektrine.DNS.Record + alias Elektrine.DNS.Tunnels alias Elektrine.DNS.Zone alias Elektrine.DNS.ZoneCache alias Elektrine.DNS.ZoneServiceConfig @@ -730,13 +731,11 @@ defmodule Elektrine.DNS do @doc """ Authorizes binding a tunnel id to a zone for proxied tunnel origins. - Until the tunnels control plane (PR 7b) is present, ownership is resolved via - `:tunnel_ownership_checker` in the `:dns` application config: - - fn tunnel_id, zone_id -> :ok | {:error, reason} end + Primary path: `Elektrine.DNS.Tunnels` control plane. Tests and temporary + overrides may install `:tunnel_ownership_checker` in the `:dns` application + config (`fn tunnel_id, zone_id -> :ok | {:error, reason} end`). Reasons: `:invalid_tunnel_id`, `:tunnel_not_found`, `:tunnel_not_owned`. - With no checker configured, authorization fails closed (`:tunnel_not_found`). """ def authorize_tunnel_for_zone(tunnel_id, zone_id) when is_binary(tunnel_id) and is_integer(zone_id) do @@ -747,7 +746,7 @@ defmodule Elektrine.DNS do normalize_tunnel_ownership_result(fun.(id, zone_id)) _ -> - {:error, :tunnel_not_found} + Tunnels.authorize_for_zone(id, zone_id) end :error -> @@ -757,6 +756,15 @@ defmodule Elektrine.DNS do def authorize_tunnel_for_zone(_, _), do: {:error, :invalid_tunnel_id} + defdelegate list_zone_tunnels(zone), to: Tunnels + defdelegate get_zone_tunnel(zone, id), to: Tunnels + def create_tunnel(zone, attrs), do: Tunnels.create_tunnel(zone, attrs, nil) + def create_tunnel(zone, attrs, created_by), do: Tunnels.create_tunnel(zone, attrs, created_by) + defdelegate update_tunnel(tunnel, attrs), to: Tunnels + defdelegate revoke_tunnel(tunnel), to: Tunnels + defdelegate public_tunnel(tunnel), to: Tunnels + def public_tunnel(tunnel, raw_token), do: Tunnels.public_tunnel(tunnel, raw_token) + def scan_existing_zone(domain) when is_binary(domain) do normalized_domain = domain |> String.trim() |> String.downcase() |> String.trim_trailing(".") diff --git a/apps/elektrine_dns/lib/elektrine/dns/tunnel.ex b/apps/elektrine_dns/lib/elektrine/dns/tunnel.ex new file mode 100644 index 0000000..48c406a --- /dev/null +++ b/apps/elektrine_dns/lib/elektrine/dns/tunnel.ex @@ -0,0 +1,185 @@ +defmodule Elektrine.DNS.Tunnel do + @moduledoc """ + Zone-owned tunnel credential for private (firewalled) HTTP origins. + + Long-lived agent tokens are stored hashed (`etn_...` shown once at mint). + Runtime WSS sessions use short-lived tickets minted at register/refresh. + """ + + use Ecto.Schema + import Ecto.Changeset + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :id + @statuses ~w(inactive active revoked) + @max_name_length 100 + @max_allowed_hosts 50 + + schema "dns_tunnels" do + field :name, :string + field :status, :string, default: "inactive" + field :allowed_hosts, {:array, :string}, default: [] + field :token_hash, :string + field :token_prefix, :string + field :last_seen_at, :utc_datetime + field :revoked_at, :utc_datetime + + # Virtual: raw token returned once on mint + field :token, :string, virtual: true + + belongs_to :zone, Elektrine.DNS.Zone + belongs_to :creator, Elektrine.Accounts.User, foreign_key: :created_by + + timestamps(type: :utc_datetime) + end + + def statuses, do: @statuses + + def changeset(tunnel, attrs) do + tunnel + |> cast(attrs, [ + :name, + :status, + :allowed_hosts, + :token_hash, + :token_prefix, + :zone_id, + :created_by, + :last_seen_at, + :revoked_at + ]) + |> update_change(:name, &normalize_name/1) + |> update_change(:allowed_hosts, &normalize_hosts/1) + |> validate_required([:name, :status, :token_hash, :token_prefix, :zone_id]) + |> validate_inclusion(:status, @statuses) + |> validate_length(:name, min: 1, max: @max_name_length) + |> validate_length(:allowed_hosts, max: @max_allowed_hosts) + |> validate_allowed_hosts() + |> unique_constraint(:token_hash) + |> unique_constraint([:zone_id, :name], name: :dns_tunnels_zone_id_name_unique) + |> foreign_key_constraint(:zone_id) + |> foreign_key_constraint(:created_by) + end + + def revoke_changeset(tunnel) do + now = DateTime.utc_now() |> DateTime.truncate(:second) + + tunnel + |> change(%{status: "revoked", revoked_at: now}) + end + + def touch_seen_changeset(tunnel) do + now = DateTime.utc_now() |> DateTime.truncate(:second) + + tunnel + |> change(%{last_seen_at: now, status: active_status(tunnel)}) + end + + def generate_token do + raw = "etn_" <> Base.url_encode64(:crypto.strong_rand_bytes(32), padding: false) + hash = hash_token(raw) + prefix = String.slice(raw, 0, 12) + {raw, hash, prefix} + end + + def hash_token(token) when is_binary(token) do + :crypto.hash(:sha256, token) |> Base.encode16(case: :lower) + end + + def revoked?(%__MODULE__{status: "revoked"}), do: true + def revoked?(%__MODULE__{revoked_at: %DateTime{}}), do: true + def revoked?(_), do: false + + def active_for_session?(%__MODULE__{} = tunnel), do: not revoked?(tunnel) + + def host_allowed?(%__MODULE__{allowed_hosts: hosts}, host) + when is_list(hosts) and is_binary(host) do + host_matches?(hosts, host) + end + + def host_allowed?(_, _), do: false + + @doc """ + Returns true if `host` is allowed by the list. + + Empty list = allow all. Entries may be exact FQDNs or a single left-most + wildcard label (`*.example.com` matches `app.example.com` but not + `example.com` or `a.b.example.com`). + """ + def host_matches?(hosts, host) when is_list(hosts) and is_binary(host) do + normalized = normalize_host(host) + + case hosts do + [] -> + true + + list -> + Enum.any?(list, fn allowed -> + host_entry_matches?(normalize_host(allowed), normalized) + end) + end + end + + def host_matches?(_, _), do: false + + defp host_entry_matches?(nil, _), do: false + defp host_entry_matches?("", _), do: false + defp host_entry_matches?(allowed, host) when allowed == host, do: true + + defp host_entry_matches?(<<"*.", rest::binary>>, host) when rest != "" do + # One label only: *.example.com matches foo.example.com, not bar.foo.example.com + case String.split(host, ".", parts: 2) do + [_label, ^rest] -> true + _ -> false + end + end + + defp host_entry_matches?(_, _), do: false + + defp active_status(%__MODULE__{status: "revoked"}), do: "revoked" + defp active_status(_), do: "active" + + defp normalize_name(nil), do: nil + + defp normalize_name(name) when is_binary(name) do + name |> String.trim() |> String.slice(0, @max_name_length) + end + + defp normalize_name(other), do: other + + defp normalize_hosts(nil), do: [] + + defp normalize_hosts(hosts) when is_list(hosts) do + hosts + |> Enum.map(&normalize_host/1) + |> Enum.reject(&(&1 in [nil, ""])) + |> Enum.uniq() + end + + defp normalize_hosts(_), do: [] + + defp normalize_host(host) when is_binary(host) do + host + |> String.trim() + |> String.trim_trailing(".") + |> String.downcase() + end + + defp normalize_host(_), do: nil + + defp validate_allowed_hosts(changeset) do + hosts = get_field(changeset, :allowed_hosts) || [] + + invalid = + Enum.reject(hosts, fn host -> + is_binary(host) and host != "" and + Regex.match?(~r/^(?:\*\.)?(?:[a-z0-9-]+\.)+[a-z]{2,}$/, host) + end) + + if invalid == [] do + changeset + else + add_error(changeset, :allowed_hosts, "contains invalid hostnames") + end + end +end diff --git a/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent.ex b/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent.ex new file mode 100644 index 0000000..1849e95 --- /dev/null +++ b/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent.ex @@ -0,0 +1,488 @@ +defmodule Elektrine.DNS.TunnelAgent do + @moduledoc """ + Outbound tunnel agent MVP. + + Registers with the control plane using a long-lived `etn_...` token, opens a + WSS session to the edge connector, and serves HTTP from a local `ORIGIN_URL` + only (never from edge-supplied free-form URLs). + + Run via: + + mix elektrine.tunnel_agent + + Environment: + + * `CONTROL_PLANE_URL` — e.g. `https://edge.example.com` + * `TUNNEL_TOKEN` — `etn_...` token from mint + * `ORIGIN_URL` — local origin base, e.g. `http://127.0.0.1:8080` + * `AGENT_VERSION` — optional version string + """ + + use GenServer + + require Logger + + alias Elektrine.DNS.TunnelAgent.WSClient + alias ElektrineWeb.Edge.TunnelFrame + + @default_backoff_ms 1_000 + @max_backoff_ms 60_000 + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def child_spec(opts) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [opts]}, + type: :worker, + restart: :permanent + } + end + + @doc """ + Run the agent until the process exits (CLI / mix task entry). + """ + def run!(opts \\ []) do + config = config_from_env(opts) + + {:ok, _pid} = start_link(config) + Process.flag(:trap_exit, true) + + receive do + {:EXIT, _pid, reason} -> + Logger.error("tunnel agent exited: #{inspect(reason)}") + System.halt(1) + end + end + + def config_from_env(opts \\ []) do + env = fn key, default -> + Keyword.get(opts, key) || System.get_env(String.upcase(to_string(key))) || default + end + + control_plane = + env.(:control_plane_url, nil) || + raise ArgumentError, "CONTROL_PLANE_URL is required" + + token = + env.(:tunnel_token, nil) || + raise ArgumentError, "TUNNEL_TOKEN is required" + + origin = + env.(:origin_url, nil) || + raise ArgumentError, "ORIGIN_URL is required" + + %{ + control_plane_url: String.trim_trailing(control_plane, "/"), + tunnel_token: token, + origin_url: String.trim_trailing(origin, "/"), + agent_version: env.(:agent_version, "elektrine-tunnel-agent/0.1.0"), + max_streams: parse_int(env.(:max_streams, "16"), 16) + } + end + + @impl true + def init(config) do + state = + Map.merge(config, %{ + ws_pid: nil, + buffer: "", + backoff_ms: @default_backoff_ms, + # stream_id => %{method, path, headers, body, has_body} + pending_streams: %{} + }) + + send(self(), :connect) + {:ok, state} + end + + @impl true + def handle_info(:connect, state) do + case register(state) do + {:ok, session} -> + case connect_ws(state, session) do + {:ok, ws_pid} -> + Logger.info("tunnel agent connected tunnel_id=#{session.tunnel_id}") + + {:noreply, + Map.merge(state, %{ + ws_pid: ws_pid, + session: session, + backoff_ms: @default_backoff_ms, + buffer: "" + })} + + {:error, reason} -> + Logger.warning("tunnel ws connect failed: #{inspect(reason)}") + schedule_reconnect(state) + end + + {:error, reason} -> + Logger.warning("tunnel register failed: #{inspect(reason)}") + schedule_reconnect(state) + end + end + + def handle_info({:ws_connected, _pid}, state), do: {:noreply, state} + + def handle_info({:ws_binary, data}, state) when is_binary(data) do + buffer = state.buffer <> data + {state, _actions} = handle_buffer(buffer, state) + {:noreply, state} + end + + def handle_info({:ws_disconnected, reason}, state) do + Logger.warning("tunnel ws disconnected: #{inspect(reason)}") + schedule_reconnect(%{state | ws_pid: nil}) + end + + def handle_info({:EXIT, pid, reason}, %{ws_pid: pid} = state) do + Logger.warning("tunnel ws process exit: #{inspect(reason)}") + schedule_reconnect(%{state | ws_pid: nil}) + end + + def handle_info(_msg, state), do: {:noreply, state} + + defp schedule_reconnect(state) do + backoff = state.backoff_ms + jitter = :rand.uniform(max(div(backoff, 4), 1)) + delay = backoff + jitter + Process.send_after(self(), :connect, delay) + next = min(backoff * 2, @max_backoff_ms) + {:noreply, %{state | backoff_ms: next}} + end + + defp register(state) do + url = state.control_plane_url <> "/_edge/tunnel/v1/register" + + headers = [ + {"authorization", "Bearer " <> state.tunnel_token}, + {"content-type", "application/json"} + ] + + body = "{}" + + case http_post(url, headers, body) do + {:ok, 200, _headers, resp_body} -> + case Jason.decode(resp_body) do + {:ok, map} -> + {:ok, + %{ + tunnel_id: map["tunnel_id"], + session_ticket: map["session_ticket"], + connector_urls: List.wrap(map["connector_urls"]), + idle_timeout_ms: map["idle_timeout_ms"] || 60_000, + max_streams: map["max_streams"] || state.max_streams + }} + + _ -> + {:error, :invalid_register_response} + end + + {:ok, status, _headers, body} -> + {:error, {:register_http, status, body}} + + {:error, reason} -> + {:error, reason} + end + end + + defp connect_ws(state, session) do + # Prefer AUTH frame for the session ticket (avoids query-string log leakage). + # Ticket is still accepted on the query string server-side for compatibility. + ws_url = + case session.connector_urls do + [url | _] when is_binary(url) and url != "" -> + strip_query_ticket(url) + + _ -> + control = state.control_plane_url + ws_base = String.replace_prefix(control, "https://", "wss://") + ws_base = String.replace_prefix(ws_base, "http://", "ws://") + ws_base <> "/_edge/tunnel/v1/ws" + end + + WSClient.start_link( + url: ws_url, + parent: self(), + session_ticket: session.session_ticket, + agent_version: state.agent_version, + max_streams: session.max_streams || state.max_streams + ) + end + + defp strip_query_ticket(url) when is_binary(url) do + case URI.parse(url) do + %URI{query: nil} = uri -> + URI.to_string(uri) + + %URI{query: query} = uri -> + cleaned = + query + |> URI.decode_query() + |> Map.drop(["session_ticket", "ticket"]) + |> URI.encode_query() + + URI.to_string(%{uri | query: if(cleaned == "", do: nil, else: cleaned)}) + end + end + + defp handle_buffer(buffer, state) do + case TunnelFrame.decode(buffer) do + {:ok, frame, rest} -> + state = handle_frame(frame, state) + handle_buffer(rest, state) + + {:incomplete, rest} -> + {%{state | buffer: rest}, []} + + {:error, reason} -> + Logger.warning("agent frame error: #{inspect(reason)}") + {%{state | buffer: ""}, []} + end + end + + defp handle_frame({:auth_ok, _payload}, state) do + Logger.info("tunnel AUTH_OK") + state + end + + defp handle_frame({:auth_fail, payload}, state) do + Logger.error("tunnel AUTH_FAIL: #{inspect(payload)}") + state + end + + defp handle_frame({:ping, opaque}, state) do + push_frame(state, :pong, opaque) + state + end + + defp handle_frame({:pong, _}, state), do: state + + defp handle_frame({:stream_open, payload}, state) do + stream_id = payload[:stream_id] || payload["stream_id"] + method = payload[:method] || payload["method"] || "GET" + path = payload[:path] || payload["path"] || "/" + headers = payload[:headers] || payload["headers"] || [] + has_body = truthy?(payload[:has_body] || payload["has_body"]) + + pending = %{ + method: method, + path: path, + headers: headers, + body: "", + has_body: has_body + } + + if has_body do + put_in(state, [:pending_streams, stream_id], pending) + else + start_origin_fetch(state, stream_id, pending) + state + end + end + + defp handle_frame({:stream_body, %{stream_id: stream_id, fin: fin, data: data}}, state) do + case Map.get(state.pending_streams, stream_id) do + nil -> + # Body without a pending open (or open had has_body=false) — ignore. + state + + pending -> + pending = %{pending | body: pending.body <> data} + + if fin do + state = update_in(state, [:pending_streams], &Map.delete(&1, stream_id)) + start_origin_fetch(state, stream_id, pending) + state + else + put_in(state, [:pending_streams, stream_id], pending) + end + end + end + + defp handle_frame({:stream_rst, payload}, state) do + Logger.debug("stream rst: #{inspect(payload)}") + state + end + + defp handle_frame(_other, state), do: state + + defp start_origin_fetch(state, stream_id, pending) do + Task.start(fn -> + serve_origin( + state, + stream_id, + pending.method, + pending.path, + pending.headers, + pending.body || "" + ) + end) + end + + defp truthy?(true), do: true + defp truthy?(1), do: true + defp truthy?("true"), do: true + defp truthy?("1"), do: true + defp truthy?(_), do: false + + defp serve_origin(state, stream_id, method, path, headers, body) do + url = state.origin_url <> ensure_leading_slash(path) + req_headers = decode_headers(headers) + + case http_request(method, url, req_headers, body) do + {:ok, status, resp_headers, resp_body} -> + push_frame(state, :stream_headers, %{ + stream_id: stream_id, + status: status, + headers: Enum.map(resp_headers, fn {k, v} -> [to_string(k), to_string(v)] end) + }) + + push_response_body(state, stream_id, resp_body || "") + + {:error, reason} -> + Logger.warning("origin fetch failed: #{inspect(reason)}") + + push_frame(state, :stream_headers, %{ + stream_id: stream_id, + status: 502, + headers: [["content-type", "text/plain"]] + }) + + push_frame(state, :stream_body, %{ + stream_id: stream_id, + fin: true, + data: "Bad Gateway" + }) + end + end + + defp push_response_body(state, stream_id, "") do + push_frame(state, :stream_body, %{stream_id: stream_id, fin: true, data: ""}) + end + + defp push_response_body(state, stream_id, resp_body) when is_binary(resp_body) do + max = TunnelFrame.stream_body_max_data_bytes() + chunks = chunk_binary(resp_body, max) + last = length(chunks) - 1 + + Enum.with_index(chunks, fn chunk, idx -> + push_frame(state, :stream_body, %{ + stream_id: stream_id, + fin: idx == last, + data: chunk + }) + end) + end + + defp push_frame(%{ws_pid: pid}, name, payload) when is_pid(pid) do + case TunnelFrame.encode(name, payload) do + {:ok, bin} -> send(pid, {:send_binary, bin}) + {:error, reason} -> Logger.warning("encode failed: #{inspect(reason)}") + end + end + + defp push_frame(_, _, _), do: :ok + + defp ensure_leading_slash(<<"/", _::binary>> = path), do: path + defp ensure_leading_slash(path), do: "/" <> path + + defp decode_headers(headers) do + Enum.flat_map(headers, fn + [k, v] -> [{to_string(k), to_string(v)}] + {k, v} -> [{to_string(k), to_string(v)}] + _ -> [] + end) + end + + defp chunk_binary(bin, _size) when bin == "", do: [""] + + defp chunk_binary(bin, size) when is_binary(bin) and size > 0 do + do_chunk(bin, size, []) + end + + defp do_chunk(<<>>, _size, acc), do: Enum.reverse(acc) + + defp do_chunk(bin, size, acc) when byte_size(bin) <= size do + Enum.reverse([bin | acc]) + end + + defp do_chunk(bin, size, acc) do + chunk = binary_part(bin, 0, size) + rest = binary_part(bin, size, byte_size(bin) - size) + do_chunk(rest, size, [chunk | acc]) + end + + defp http_post(url, headers, body) do + http_request("POST", url, headers, body) + end + + defp http_request(method, url, headers, body) do + # Prefer Finch-free stdlib httpc for the agent surface. + :inets.start() + :ssl.start() + + method_atom = + method + |> to_string() + |> String.downcase() + |> String.to_atom() + + headers_charlist = + Enum.map(headers, fn {k, v} -> + {:erlang.binary_to_list(to_string(k)), :erlang.binary_to_list(to_string(v))} + end) + + request = + case method_atom do + m when m in [:post, :put, :patch] -> + content_type = + headers + |> Enum.find_value("application/octet-stream", fn + {k, v} -> if String.downcase(to_string(k)) == "content-type", do: to_string(v) + end) + + { + String.to_charlist(url), + headers_charlist, + String.to_charlist(content_type), + body || "" + } + + _ -> + {String.to_charlist(url), headers_charlist} + end + + # Do not follow redirects — a local origin 302 to an off-host URL would + # expand the agent SSRF surface (e.g. link-local metadata endpoints). + http_opts = [timeout: 30_000, connect_timeout: 10_000, autoredirect: false] + opts = [body_format: :binary] + + case :httpc.request(method_atom, request, http_opts, opts) do + {:ok, {{_http, status, _}, resp_headers, resp_body}} -> + headers = + Enum.map(resp_headers, fn {k, v} -> + {List.to_string(k), List.to_string(v)} + end) + + body = if is_binary(resp_body), do: resp_body, else: IO.iodata_to_binary(resp_body) + {:ok, status, headers, body} + + {:error, reason} -> + {:error, reason} + end + end + + defp parse_int(value, default) when is_binary(value) do + case Integer.parse(value) do + {int, ""} when int > 0 -> int + _ -> default + end + end + + defp parse_int(value, _default) when is_integer(value) and value > 0, do: value + defp parse_int(_, default), do: default +end diff --git a/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent/ws_client.ex b/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent/ws_client.ex new file mode 100644 index 0000000..bc595b3 --- /dev/null +++ b/apps/elektrine_dns/lib/elektrine/dns/tunnel_agent/ws_client.ex @@ -0,0 +1,70 @@ +defmodule Elektrine.DNS.TunnelAgent.WSClient do + @moduledoc false + + use WebSockex + + require Logger + + alias ElektrineWeb.Edge.TunnelFrame + + def start_link(opts) do + url = Keyword.fetch!(opts, :url) + parent = Keyword.fetch!(opts, :parent) + + state = %{ + parent: parent, + session_ticket: Keyword.get(opts, :session_ticket), + agent_version: Keyword.get(opts, :agent_version, "elektrine-tunnel-agent/0.1.0"), + max_streams: Keyword.get(opts, :max_streams, 16), + authed: false + } + + WebSockex.start_link(url, __MODULE__, state, name: nil) + end + + @impl true + def handle_connect(_conn, state) do + send(state.parent, {:ws_connected, self()}) + + # AUTH frame if ticket available (also may be in query string server-side). + if is_binary(state.session_ticket) and state.session_ticket != "" do + frame = + TunnelFrame.encode!(:auth, %{ + session_ticket: state.session_ticket, + agent_version: state.agent_version, + max_streams: state.max_streams + }) + + {:reply, {:binary, frame}, state} + else + {:ok, state} + end + end + + @impl true + def handle_frame({:binary, data}, state) do + send(state.parent, {:ws_binary, data}) + {:ok, state} + end + + def handle_frame(_frame, state), do: {:ok, state} + + @impl true + def handle_info({:send_binary, data}, state) when is_binary(data) do + {:reply, {:binary, data}, state} + end + + def handle_info(_msg, state), do: {:ok, state} + + @impl true + def handle_disconnect(%{reason: reason}, state) do + send(state.parent, {:ws_disconnected, reason}) + {:ok, state} + end + + @impl true + def terminate(reason, state) do + send(state.parent, {:ws_disconnected, reason}) + :ok + end +end diff --git a/apps/elektrine_dns/lib/elektrine/dns/tunnels.ex b/apps/elektrine_dns/lib/elektrine/dns/tunnels.ex new file mode 100644 index 0000000..346af3a --- /dev/null +++ b/apps/elektrine_dns/lib/elektrine/dns/tunnels.ex @@ -0,0 +1,249 @@ +defmodule Elektrine.DNS.Tunnels do + @moduledoc """ + Control-plane operations for DNS edge tunnels. + + Mint/revoke are zone-owner operations (external API scope `write:edge`). + Agents exchange long-lived `etn_...` tokens for short-lived session tickets + via `/_edge/tunnel/v1/*` when `DNS_TUNNEL_ENABLED=true`. + """ + + import Ecto.Query, warn: false + + alias Elektrine.DNS.Tunnel + alias Elektrine.DNS.Zone + alias Elektrine.Repo + + @default_max_per_zone 10 + @session_ttl_seconds 3_600 + + @doc """ + Returns true when the tunnel control plane is enabled. + """ + def enabled? do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_enabled, false) + end + + def max_per_zone do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_max_per_zone, @default_max_per_zone) + end + + def max_streams do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_max_streams, 16) + end + + def idle_timeout_ms do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_idle_timeout_ms, 60_000) + end + + def stream_idle_timeout_ms do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_stream_idle_timeout_ms, 120_000) + end + + def session_ttl_seconds do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_session_ttl_seconds, @session_ttl_seconds) + end + + def connector_urls do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_connector_urls, []) + end + + @doc """ + Ownership checker used by `Elektrine.DNS.authorize_tunnel_for_zone/2`. + """ + def authorize_for_zone(tunnel_id, zone_id) + when is_binary(tunnel_id) and is_integer(zone_id) do + case get_tunnel(tunnel_id) do + %Tunnel{zone_id: ^zone_id} = tunnel -> + if Tunnel.revoked?(tunnel), do: {:error, :tunnel_not_owned}, else: :ok + + %Tunnel{} -> + {:error, :tunnel_not_owned} + + nil -> + {:error, :tunnel_not_found} + end + end + + def authorize_for_zone(_, _), do: {:error, :invalid_tunnel_id} + + def list_zone_tunnels(%Zone{id: zone_id}), do: list_zone_tunnels(zone_id) + + def list_zone_tunnels(zone_id) when is_integer(zone_id) do + Tunnel + |> where([t], t.zone_id == ^zone_id and t.status != "revoked") + |> order_by([t], asc: t.name) + |> Repo.all() + end + + def list_zone_tunnels(_), do: [] + + def get_tunnel(id) when is_binary(id) do + case Ecto.UUID.cast(id) do + {:ok, uuid} -> Repo.get(Tunnel, uuid) + :error -> nil + end + end + + def get_tunnel(_), do: nil + + def get_zone_tunnel(%Zone{id: zone_id}, id), do: get_zone_tunnel(zone_id, id) + + def get_zone_tunnel(zone_id, id) when is_integer(zone_id) and is_binary(id) do + case get_tunnel(id) do + %Tunnel{zone_id: ^zone_id} = tunnel -> tunnel + _ -> nil + end + end + + def get_zone_tunnel(_, _), do: nil + + @doc """ + Mint a new tunnel for the zone. Returns `{:ok, tunnel}` with virtual `:token` + set once. Requires tunnels to be enabled for the control plane to accept + agent register calls; minting itself is allowed whenever the table exists + so operators can pre-provision while the flag is off. + """ + def create_tunnel(%Zone{} = zone, attrs, created_by \\ nil) do + with :ok <- enforce_max_tunnels(zone.id) do + {raw, hash, prefix} = Tunnel.generate_token() + + attrs = + attrs + |> stringify_keys() + |> Map.put("token_hash", hash) + |> Map.put("token_prefix", prefix) + |> Map.put("zone_id", zone.id) + |> Map.put("status", "inactive") + |> maybe_put_created_by(created_by) + + %Tunnel{} + |> Tunnel.changeset(attrs) + |> Repo.insert() + |> case do + {:ok, tunnel} -> {:ok, %{tunnel | token: raw}} + other -> other + end + end + end + + def update_tunnel(%Tunnel{} = tunnel, attrs) when is_map(attrs) do + if Tunnel.revoked?(tunnel) do + {:error, :revoked} + else + tunnel + |> Tunnel.changeset(stringify_keys(attrs)) + |> Repo.update() + end + end + + def revoke_tunnel(%Tunnel{} = tunnel) do + case tunnel |> Tunnel.revoke_changeset() |> Repo.update() do + {:ok, revoked} = ok -> + # Drop live WSS sessions and ETS tickets on this node (best-effort; + # multi-node edges need sticky register/ws or fan-out later). + disconnect_live_session(revoked.id) + ok + + other -> + other + end + end + + defp disconnect_live_session(tunnel_id) when is_binary(tunnel_id) do + connector = ElektrineWeb.Edge.Connector + + if Code.ensure_loaded?(connector) and function_exported?(connector, :disconnect_tunnel, 1) do + try do + connector.disconnect_tunnel(tunnel_id) + catch + :exit, _ -> :ok + end + else + :ok + end + end + + @doc """ + Resolve a long-lived agent token to a tunnel row. + """ + def get_tunnel_by_token(raw_token) when is_binary(raw_token) do + hash = Tunnel.hash_token(String.trim(raw_token)) + + case Repo.get_by(Tunnel, token_hash: hash) do + %Tunnel{} = tunnel -> + if Tunnel.revoked?(tunnel), do: {:error, :revoked}, else: {:ok, tunnel} + + nil -> + {:error, :not_found} + end + end + + def get_tunnel_by_token(_), do: {:error, :not_found} + + def touch_last_seen(%Tunnel{} = tunnel) do + tunnel + |> Tunnel.touch_seen_changeset() + |> Repo.update() + end + + @doc """ + Public representation without secrets. + """ + def public_tunnel(%Tunnel{} = tunnel) do + %{ + id: tunnel.id, + zone_id: tunnel.zone_id, + name: tunnel.name, + status: tunnel.status, + allowed_hosts: tunnel.allowed_hosts || [], + token_prefix: tunnel.token_prefix, + last_seen_at: tunnel.last_seen_at, + revoked_at: tunnel.revoked_at, + inserted_at: tunnel.inserted_at, + updated_at: tunnel.updated_at + } + end + + def public_tunnel(%Tunnel{} = tunnel, raw_token) when is_binary(raw_token) do + tunnel + |> public_tunnel() + |> Map.put(:token, raw_token) + end + + defp enforce_max_tunnels(zone_id) do + count = + Tunnel + |> where([t], t.zone_id == ^zone_id and t.status != "revoked") + |> Repo.aggregate(:count, :id) + + if count >= max_per_zone() do + {:error, :max_tunnels_reached} + else + :ok + end + end + + defp maybe_put_created_by(attrs, %{id: user_id}) when is_integer(user_id) do + Map.put(attrs, "created_by", user_id) + end + + defp maybe_put_created_by(attrs, user_id) when is_integer(user_id) do + Map.put(attrs, "created_by", user_id) + end + + defp maybe_put_created_by(attrs, _), do: attrs + + defp stringify_keys(attrs) when is_map(attrs) do + Map.new(attrs, fn + {k, v} when is_atom(k) -> {Atom.to_string(k), v} + {k, v} -> {k, v} + end) + end +end diff --git a/apps/elektrine_dns/lib/elektrine/dns/zone.ex b/apps/elektrine_dns/lib/elektrine/dns/zone.ex index b07dc70..3d3bd3a 100644 --- a/apps/elektrine_dns/lib/elektrine/dns/zone.ex +++ b/apps/elektrine_dns/lib/elektrine/dns/zone.ex @@ -38,6 +38,7 @@ field :axfr_enabled, :boolean, default: false has_many :edge_rules, Elektrine.DNS.EdgeRule, foreign_key: :zone_id has_many :dnssec_keys, Elektrine.DNS.DnssecKey, foreign_key: :zone_id has_many :tsig_keys, Elektrine.DNS.TsigKey, foreign_key: :zone_id +has_many :tunnels, Elektrine.DNS.Tunnel, foreign_key: :zone_id timestamps(type: :utc_datetime) end diff --git a/apps/elektrine_dns/lib/elektrine_dns_web/controllers/api/dns_controller.ex b/apps/elektrine_dns/lib/elektrine_dns_web/controllers/api/dns_controller.ex index 43936c7..cda156c 100644 --- a/apps/elektrine_dns/lib/elektrine_dns_web/controllers/api/dns_controller.ex +++ b/apps/elektrine_dns/lib/elektrine_dns_web/controllers/api/dns_controller.ex @@ -8,11 +8,138 @@ defmodule ElektrineDNSWeb.API.DNSController do alias Elektrine.DNS alias Elektrine.DNS.EdgeRule alias Elektrine.DNS.EdgeRules +alias Elektrine.DNS.Tunnel alias Elektrine.DNS.Zone alias ElektrineWeb.API.Response action_fallback ElektrineWeb.FallbackController + # --- Edge tunnels (read:edge / write:edge) -------------------------------- + + def list_tunnels(conn, %{"zone_id" => zone_id}) do + user = conn.assigns.current_user + + with {:ok, id} <- parse_id(zone_id), + %Zone{} = zone <- DNS.get_zone(id, user.id) do + tunnels = Enum.map(DNS.list_zone_tunnels(zone), &DNS.public_tunnel/1) + Response.ok(conn, %{tunnels: tunnels}) + else + {:error, :bad_request} -> + Response.error(conn, :bad_request, "invalid_id", "Invalid zone id") + + nil -> + Response.error(conn, :not_found, "not_found", "Zone not found") + end + end + + def show_tunnel(conn, %{"zone_id" => zone_id, "id" => id}) do + user = conn.assigns.current_user + + with {:ok, zid} <- parse_id(zone_id), + %Zone{} = zone <- DNS.get_zone(zid, user.id), + %Tunnel{} = tunnel <- DNS.get_zone_tunnel(zone, id) do + Response.ok(conn, %{tunnel: DNS.public_tunnel(tunnel)}) + else + {:error, :bad_request} -> + Response.error(conn, :bad_request, "invalid_id", "Invalid zone id") + + nil -> + Response.error(conn, :not_found, "not_found", "Tunnel or zone not found") + end + end + + def create_tunnel(conn, %{"zone_id" => zone_id} = params) do + user = conn.assigns.current_user + attrs = Map.get(params, "tunnel", params) + + with {:ok, zid} <- parse_id(zone_id), + %Zone{} = zone <- DNS.get_zone(zid, user.id) do + case DNS.create_tunnel(zone, attrs, user) do + {:ok, tunnel} -> + Response.created(conn, %{tunnel: DNS.public_tunnel(tunnel, tunnel.token)}) + + {:error, :max_tunnels_reached} -> + Response.error( + conn, + :unprocessable_entity, + "max_tunnels_reached", + "Maximum tunnels for this zone reached" + ) + + {:error, changeset} -> + Response.error( + conn, + :unprocessable_entity, + "validation_failed", + "Invalid tunnel", + errors_on(changeset) + ) + end + else + {:error, :bad_request} -> + Response.error(conn, :bad_request, "invalid_id", "Invalid zone id") + + nil -> + Response.error(conn, :not_found, "not_found", "Zone not found") + end + end + + def update_tunnel(conn, %{"zone_id" => zone_id, "id" => id} = params) do + user = conn.assigns.current_user + attrs = Map.get(params, "tunnel", params) + + with {:ok, zid} <- parse_id(zone_id), + %Zone{} = zone <- DNS.get_zone(zid, user.id), + %Tunnel{} = tunnel <- DNS.get_zone_tunnel(zone, id), + {:ok, tunnel} <- DNS.update_tunnel(tunnel, attrs) do + Response.ok(conn, %{tunnel: DNS.public_tunnel(tunnel)}) + else + {:error, :bad_request} -> + Response.error(conn, :bad_request, "invalid_id", "Invalid zone id") + + nil -> + Response.error(conn, :not_found, "not_found", "Tunnel or zone not found") + + {:error, :revoked} -> + Response.error(conn, :unprocessable_entity, "revoked", "Tunnel is revoked") + + {:error, changeset} -> + Response.error( + conn, + :unprocessable_entity, + "validation_failed", + "Invalid tunnel", + errors_on(changeset) + ) + end + end + + def revoke_tunnel(conn, %{"zone_id" => zone_id, "id" => id}) do + user = conn.assigns.current_user + + with {:ok, zid} <- parse_id(zone_id), + %Zone{} = zone <- DNS.get_zone(zid, user.id), + %Tunnel{} = tunnel <- DNS.get_zone_tunnel(zone, id), + {:ok, _tunnel} <- DNS.revoke_tunnel(tunnel) do + Response.ok(conn, %{message: "Tunnel revoked"}) + else + {:error, :bad_request} -> + Response.error(conn, :bad_request, "invalid_id", "Invalid zone id") + + nil -> + Response.error(conn, :not_found, "not_found", "Tunnel or zone not found") + + {:error, changeset} -> + Response.error( + conn, + :unprocessable_entity, + "validation_failed", + "Could not revoke tunnel", + errors_on(changeset) + ) + end + end + def index(conn, _params) do user = conn.assigns.current_user diff --git a/apps/elektrine_dns/lib/mix/tasks/elektrine.tunnel_agent.ex b/apps/elektrine_dns/lib/mix/tasks/elektrine.tunnel_agent.ex new file mode 100644 index 0000000..fa4a2fb --- /dev/null +++ b/apps/elektrine_dns/lib/mix/tasks/elektrine.tunnel_agent.ex @@ -0,0 +1,25 @@ +defmodule Mix.Tasks.Elektrine.TunnelAgent do + @moduledoc """ + Run the Elektrine edge tunnel agent (MVP). + + CONTROL_PLANE_URL=https://edge.example.com \\ + TUNNEL_TOKEN=etn_... \\ + ORIGIN_URL=http://127.0.0.1:8080 \\ + mix elektrine.tunnel_agent + + The agent publishes only the local ORIGIN_URL; edge frames never supply a free + origin URL (SSRF boundary). + """ + + use Mix.Task + + @shortdoc "Run the DNS edge tunnel agent" + + @impl Mix.Task + def run(_args) do + Mix.Task.run("app.start") + + # Keep the VM alive for the agent loop. + Elektrine.DNS.TunnelAgent.run!() + end +end diff --git a/apps/elektrine_dns/mix.exs b/apps/elektrine_dns/mix.exs index 3264d6b..4bc10b2 100644 --- a/apps/elektrine_dns/mix.exs +++ b/apps/elektrine_dns/mix.exs @@ -29,7 +29,9 @@ defmodule ElektrineDNS.MixProject do {:phoenix, "== 1.8.9"}, {:phoenix_html, "== 4.3.0"}, {:phoenix_live_view, "== 1.1.30"}, - {:jason, "== 1.4.5"} + {:jason, "== 1.4.5"}, + # WSS client for the outbound tunnel agent (KD-5b). Pinned explicitly. + {:websockex, "== 0.5.1"} ] end diff --git a/apps/elektrine_dns/test/elektrine/dns/tunnel_agent_test.exs b/apps/elektrine_dns/test/elektrine/dns/tunnel_agent_test.exs new file mode 100644 index 0000000..21a5868 --- /dev/null +++ b/apps/elektrine_dns/test/elektrine/dns/tunnel_agent_test.exs @@ -0,0 +1,143 @@ +defmodule Elektrine.DNS.TunnelAgentTest do + use ExUnit.Case, async: true + + alias Elektrine.DNS.TunnelAgent + alias ElektrineWeb.Edge.TunnelFrame + + # Drive private frame handling by reusing the GenServer's frame path via a + # thin test double: start the agent without connecting (mock connect loop). + + test "buffers POST body until STREAM_BODY fin before origin fetch" do + # Pure-style exercise: use the agent module's public frame helpers through + # a temporary GenServer that does not dial the network. + {:ok, pid} = + GenServer.start_link(__MODULE__.Harness, %{ + buffer: "", + pending_streams: %{}, + origin_calls: self() + }) + + open = + TunnelFrame.encode!(:stream_open, %{ + stream_id: 1, + method: "POST", + path: "/echo", + headers: [["content-type", "text/plain"]], + has_body: true + }) + + body1 = + TunnelFrame.encode!(:stream_body, %{stream_id: 1, fin: false, data: "hel"}) + + body2 = + TunnelFrame.encode!(:stream_body, %{stream_id: 1, fin: true, data: "lo"}) + + send(pid, {:ws_binary, open <> body1}) + refute_receive {:origin_fetch, _}, 50 + + send(pid, {:ws_binary, body2}) + assert_receive {:origin_fetch, %{method: "POST", path: "/echo", body: "hello"}}, 500 + + GenServer.stop(pid) + end + + test "GET without body starts origin fetch immediately" do + {:ok, pid} = + GenServer.start_link(__MODULE__.Harness, %{ + buffer: "", + pending_streams: %{}, + origin_calls: self() + }) + + open = + TunnelFrame.encode!(:stream_open, %{ + stream_id: 3, + method: "GET", + path: "/health", + headers: [], + has_body: false + }) + + send(pid, {:ws_binary, open}) + assert_receive {:origin_fetch, %{method: "GET", path: "/health", body: ""}}, 500 + GenServer.stop(pid) + end + + defmodule Harness do + @moduledoc false + use GenServer + + alias ElektrineWeb.Edge.TunnelFrame + + def start_link(state), do: GenServer.start_link(__MODULE__, state) + + @impl true + def init(state), do: {:ok, state} + + @impl true + def handle_info({:ws_binary, data}, state) do + buffer = Map.get(state, :buffer, "") <> data + {state, _} = handle_buffer(buffer, state) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} + + defp handle_buffer(buffer, state) do + case TunnelFrame.decode(buffer) do + {:ok, frame, rest} -> + state = handle_frame(frame, state) + handle_buffer(rest, state) + + {:incomplete, rest} -> + {%{state | buffer: rest}, []} + + {:error, _} -> + {%{state | buffer: ""}, []} + end + end + + # Mirror TunnelAgent stream open/body handling (keep in sync). + defp handle_frame({:stream_open, payload}, state) do + stream_id = payload[:stream_id] || payload["stream_id"] + method = payload[:method] || payload["method"] || "GET" + path = payload[:path] || payload["path"] || "/" + headers = payload[:headers] || payload["headers"] || [] + has_body = payload[:has_body] == true or payload["has_body"] == true + + pending = %{method: method, path: path, headers: headers, body: "", has_body: has_body} + + if has_body do + put_in(state, [:pending_streams, stream_id], pending) + else + send(state.origin_calls, {:origin_fetch, pending}) + state + end + end + + defp handle_frame({:stream_body, %{stream_id: stream_id, fin: fin, data: data}}, state) do + case Map.get(state.pending_streams, stream_id) do + nil -> + state + + pending -> + pending = %{pending | body: pending.body <> data} + + if fin do + state = update_in(state, [:pending_streams], &Map.delete(&1, stream_id)) + send(state.origin_calls, {:origin_fetch, pending}) + state + else + put_in(state, [:pending_streams, stream_id], pending) + end + end + end + + defp handle_frame(_other, state), do: state + end + + # Silence unused alias warning if TunnelAgent is only referenced in docs. + test "agent module is loadable" do + assert Code.ensure_loaded?(TunnelAgent) + end +end diff --git a/apps/elektrine_dns/test/elektrine/dns/tunnels_test.exs b/apps/elektrine_dns/test/elektrine/dns/tunnels_test.exs new file mode 100644 index 0000000..b8bed19 --- /dev/null +++ b/apps/elektrine_dns/test/elektrine/dns/tunnels_test.exs @@ -0,0 +1,119 @@ +defmodule Elektrine.DNS.TunnelsTest do + use Elektrine.DataCase, async: false + + alias Elektrine.AccountsFixtures + alias Elektrine.DNS + alias Elektrine.DNS.Tunnel + alias Elektrine.DNS.Tunnels + + setup do + user = AccountsFixtures.user_fixture() + {:ok, zone} = DNS.create_zone(user, %{"domain" => unique_domain()}) + %{user: user, zone: zone} + end + + test "mints tunnel with etn_ token shown once", %{user: user, zone: zone} do + assert {:ok, tunnel} = + Tunnels.create_tunnel(zone, %{"name" => "lab", "allowed_hosts" => []}, user) + + assert String.starts_with?(tunnel.token, "etn_") + assert tunnel.token_prefix == String.slice(tunnel.token, 0, 12) + assert tunnel.status == "inactive" + assert tunnel.zone_id == zone.id + + # Token hash is stored; raw not re-readable + reloaded = Tunnels.get_tunnel(tunnel.id) + assert reloaded.token_hash == Tunnel.hash_token(tunnel.token) + refute Map.get(reloaded, :token) + + assert {:ok, ^reloaded} = Tunnels.get_tunnel_by_token(tunnel.token) + end + + test "authorize_for_zone checks ownership and revocation", %{user: user, zone: zone} do + other = AccountsFixtures.user_fixture() + {:ok, other_zone} = DNS.create_zone(other, %{"domain" => unique_domain()}) + + assert {:ok, tunnel} = Tunnels.create_tunnel(zone, %{"name" => "a"}, user) + assert :ok = Tunnels.authorize_for_zone(tunnel.id, zone.id) + assert {:error, :tunnel_not_owned} = Tunnels.authorize_for_zone(tunnel.id, other_zone.id) + + assert {:ok, revoked} = Tunnels.revoke_tunnel(tunnel) + assert Tunnel.revoked?(revoked) + assert {:error, :tunnel_not_owned} = Tunnels.authorize_for_zone(tunnel.id, zone.id) + end + + test "DNS.authorize_tunnel_for_zone uses control plane by default", %{user: user, zone: zone} do + assert {:ok, tunnel} = Tunnels.create_tunnel(zone, %{"name" => "b"}, user) + assert :ok = DNS.authorize_tunnel_for_zone(tunnel.id, zone.id) + + assert {:error, :tunnel_not_found} = + DNS.authorize_tunnel_for_zone(Ecto.UUID.generate(), zone.id) + end + + test "enforces max tunnels per zone", %{user: user, zone: zone} do + previous = Application.get_env(:elektrine, :dns, []) + + Application.put_env( + :elektrine, + :dns, + Keyword.put(previous, :tunnel_max_per_zone, 1) + ) + + on_exit(fn -> Application.put_env(:elektrine, :dns, previous) end) + + assert {:ok, _} = Tunnels.create_tunnel(zone, %{"name" => "one"}, user) + assert {:error, :max_tunnels_reached} = Tunnels.create_tunnel(zone, %{"name" => "two"}, user) + end + + test "host_allowed? empty list allows any host" do + tunnel = %Tunnel{allowed_hosts: []} + assert Tunnel.host_allowed?(tunnel, "app.example.com") + end + + test "host_allowed? restricts to list" do + tunnel = %Tunnel{allowed_hosts: ["app.example.com"]} + assert Tunnel.host_allowed?(tunnel, "app.example.com") + refute Tunnel.host_allowed?(tunnel, "other.example.com") + end + + test "host_matches? supports single-label wildcards" do + assert Tunnel.host_matches?(["*.example.com"], "app.example.com") + assert Tunnel.host_matches?(["*.example.com"], "APP.Example.COM") + refute Tunnel.host_matches?(["*.example.com"], "example.com") + refute Tunnel.host_matches?(["*.example.com"], "a.b.example.com") + refute Tunnel.host_matches?(["*.example.com"], "other.test") + assert Tunnel.host_matches?(["app.example.com", "*.cdn.test"], "img.cdn.test") + end + + test "revoke disconnects live connector session and tickets", %{user: user, zone: zone} do + assert {:ok, tunnel} = Tunnels.create_tunnel(zone, %{"name" => "live"}, user) + + # Start connector if needed and mint + register a fake session. + ensure_connector() + assert {:ok, ticket, _} = ElektrineWeb.Edge.Connector.mint_session_ticket(tunnel) + assert {:ok, _} = ElektrineWeb.Edge.Connector.authenticate_ticket(ticket) + + agent = spawn(fn -> Process.sleep(60_000) end) + :ok = ElektrineWeb.Edge.Connector.register_ws(tunnel.id, agent, %{}) + assert ElektrineWeb.Edge.Connector.session_online?(tunnel.id) + + assert {:ok, _} = Tunnels.revoke_tunnel(tunnel) + refute ElektrineWeb.Edge.Connector.session_online?(tunnel.id) + assert {:error, :invalid_ticket} = ElektrineWeb.Edge.Connector.authenticate_ticket(ticket) + end + + defp ensure_connector do + case Process.whereis(ElektrineWeb.Edge.Connector) do + nil -> + {:ok, _pid} = start_supervised(ElektrineWeb.Edge.Connector) + :ok + + _ -> + :ok + end + end + + defp unique_domain do + "tunnels-#{System.unique_integer([:positive])}.example.com" + end +end diff --git a/apps/elektrine_social/lib/elektrine_social_web/live/timeline_live/operations/post_operations.ex b/apps/elektrine_social/lib/elektrine_social_web/live/timeline_live/operations/post_operations.ex index d05c2fb..3728d8d 100644 --- a/apps/elektrine_social/lib/elektrine_social_web/live/timeline_live/operations/post_operations.ex +++ b/apps/elektrine_social/lib/elektrine_social_web/live/timeline_live/operations/post_operations.ex @@ -121,6 +121,7 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do title = params["title"] || socket.assigns.new_post_title cw = params["cw"] scheduled_at_input = params["scheduled_at"] || socket.assigns[:new_post_scheduled_at] || "" + visibility = params["visibility"] || socket.assigns.new_post_visibility || "public" @@ -467,6 +468,7 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do # Prefer form params (button may submit the form) so the latest textarea value is used content = params["content"] || socket.assigns.new_post_content || "" title = params["title"] || socket.assigns.new_post_title + visibility = params["visibility"] || socket.assigns.new_post_visibility || "public" @@ -487,7 +489,8 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do media_metadata: media_metadata, alt_texts: alt_texts, content_warning: content_warning, - sensitive: socket.assigns.new_post_sensitive || Elektrine.Strings.present?(content_warning) + sensitive: + socket.assigns.new_post_sensitive || Elektrine.Strings.present?(content_warning) ] opts = @@ -530,7 +533,11 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do Logger.warning("save_draft failed for user #{user.id}: #{inspect(reason)}") {:noreply, - put_flash(socket, :error, Elektrine.UserError.flash("Failed to save draft", reason))} + put_flash( + socket, + :error, + Elektrine.UserError.flash("Failed to save draft", reason) + )} end {:error, message} -> @@ -561,7 +568,11 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do {:error, reason} -> {:noreply, - put_flash(socket, :error, Elektrine.UserError.flash("Failed to save draft", reason))} + put_flash( + socket, + :error, + Elektrine.UserError.flash("Failed to save draft", reason) + )} end end else @@ -709,6 +720,79 @@ defmodule ElektrineSocialWeb.TimelineLive.Operations.PostOperations do end end + defp do_create_post(content, params, socket) do + visibility = + params["visibility"] || socket.assigns.new_post_visibility || "public" + + user = socket.assigns.current_user + has_content = Elektrine.Strings.present?(content) + has_attachments = !Enum.empty?(socket.assigns.pending_media_urls) + + if !has_content && !has_attachments do + {:noreply, put_flash(socket, :error, "Post cannot be empty")} + else + title = + case params["title"] do + nil -> nil + "" -> nil + t -> String.trim(t) + end + + uploaded_files = socket.assigns.pending_media_urls + media_metadata = pending_media_metadata(socket) + alt_texts = socket.assigns.pending_media_alt_texts || %{} + scheduled_at_input = params["scheduled_at"] || socket.assigns[:new_post_scheduled_at] || "" + post_opts = [visibility: visibility, media_urls: uploaded_files] + + post_opts = + if map_size(media_metadata) == 0 do + post_opts + else + Keyword.put(post_opts, :media_metadata, media_metadata) + end + + post_opts = + if title do + Keyword.put(post_opts, :title, title) + else + post_opts + end + + post_opts = + if Enum.empty?(alt_texts) do + post_opts + else + Keyword.put(post_opts, :alt_texts, alt_texts) + end + + post_opts = + if socket.assigns.new_post_content_warning do + Keyword.put(post_opts, :content_warning, socket.assigns.new_post_content_warning) + else + post_opts + end + + post_opts = + if socket.assigns.new_post_sensitive do + Keyword.put(post_opts, :sensitive, true) + else + post_opts + end + + case parse_schedule_input(scheduled_at_input) do + {:ok, scheduled_at} -> + if scheduled_at do + schedule_post(socket, user, content, post_opts, scheduled_at) + else + publish_post_now(socket, user, content, post_opts) + end + + {:error, message} -> + {:noreply, put_flash(socket, :error, message)} + end + end + end + def handle_load_more(socket) do case allow_timeline_read(socket, :load_more) do :ok -> diff --git a/apps/elektrine_web/lib/elektrine_web/controllers/tunnel_controller.ex b/apps/elektrine_web/lib/elektrine_web/controllers/tunnel_controller.ex new file mode 100644 index 0000000..6615dee --- /dev/null +++ b/apps/elektrine_web/lib/elektrine_web/controllers/tunnel_controller.ex @@ -0,0 +1,202 @@ +defmodule ElektrineWeb.TunnelController do + @moduledoc """ + Internal tunnel agent control-plane endpoints under `/_edge/tunnel/v1`. + + Auth: long-lived tunnel token (`Authorization: Bearer etn_...`) for + register/refresh/disconnect. WebSocket upgrade is handled separately. + """ + + use ElektrineWeb, :controller + + require Logger + + alias ElektrineWeb.Edge.Connector + + def register(conn, _params) do + with :ok <- require_enabled(), + {:ok, tunnel} <- tunnel_from_conn(conn), + {:ok, ticket, expires_at} <- Connector.mint_session_ticket(tunnel) do + _ = touch_last_seen(tunnel) + + json(conn, %{ + tunnel_id: tunnel.id, + session_ticket: ticket, + expires_at: expires_at, + connector_urls: connector_urls(conn), + idle_timeout_ms: tunnel_config(:tunnel_idle_timeout_ms, 60_000), + stream_idle_timeout_ms: tunnel_config(:tunnel_stream_idle_timeout_ms, 120_000), + max_streams: tunnel_config(:tunnel_max_streams, 16) + }) + else + {:error, :disabled} -> + disabled(conn) + + {:error, :missing_token} -> + unauthorized(conn, "missing_token") + + {:error, :not_found} -> + unauthorized(conn, "invalid_token") + + {:error, :revoked} -> + unauthorized(conn, "revoked") + + {:error, reason} -> + Logger.warning("tunnel register failed: #{inspect(reason)}") + error(conn, :internal_server_error, "register_failed") + end + end + + def refresh(conn, _params) do + with :ok <- require_enabled(), + {:ok, tunnel} <- tunnel_from_conn(conn), + {:ok, ticket, expires_at} <- Connector.mint_session_ticket(tunnel) do + _ = touch_last_seen(tunnel) + + json(conn, %{ + tunnel_id: tunnel.id, + session_ticket: ticket, + expires_at: expires_at, + connector_urls: connector_urls(conn), + idle_timeout_ms: tunnel_config(:tunnel_idle_timeout_ms, 60_000), + max_streams: tunnel_config(:tunnel_max_streams, 16) + }) + else + {:error, :disabled} -> disabled(conn) + {:error, :missing_token} -> unauthorized(conn, "missing_token") + {:error, :not_found} -> unauthorized(conn, "invalid_token") + {:error, :revoked} -> unauthorized(conn, "revoked") + {:error, _} -> error(conn, :internal_server_error, "refresh_failed") + end + end + + def disconnect(conn, _params) do + with :ok <- require_enabled(), + {:ok, tunnel} <- tunnel_from_conn(conn) do + :ok = Connector.disconnect_tunnel(tunnel.id) + json(conn, %{ok: true, tunnel_id: tunnel.id}) + else + {:error, :disabled} -> disabled(conn) + {:error, :missing_token} -> unauthorized(conn, "missing_token") + {:error, :not_found} -> unauthorized(conn, "invalid_token") + {:error, :revoked} -> unauthorized(conn, "revoked") + {:error, _} -> error(conn, :internal_server_error, "disconnect_failed") + end + end + + @doc """ + Upgrade the connection to the tunnel WebSock handler. + """ + def ws(conn, params) do + case require_enabled() do + :ok -> + ticket = + Map.get(params, "session_ticket") || + Map.get(params, "ticket") || + first_header(conn, "x-tunnel-session-ticket") + + opts = %{session_ticket: ticket} + idle = tunnel_config(:tunnel_idle_timeout_ms, 60_000) + + conn + |> WebSockAdapter.upgrade(ElektrineWeb.Edge.TunnelWebSock, opts, timeout: idle * 2) + |> halt() + + {:error, :disabled} -> + disabled(conn) + end + end + + defp require_enabled do + enabled? = + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_enabled, false) + + if enabled? and connector_running?(), do: :ok, else: {:error, :disabled} + end + + defp connector_running? do + case Process.whereis(Connector) do + pid when is_pid(pid) -> true + _ -> false + end + end + + defp tunnel_from_conn(conn) do + case bearer_token(conn) do + nil -> + {:error, :missing_token} + + token -> + tunnels_mod().get_tunnel_by_token(token) + end + end + + defp touch_last_seen(tunnel) do + tunnels_mod().touch_last_seen(tunnel) + end + + defp tunnels_mod, do: Elektrine.DNS.Tunnels + + defp bearer_token(conn) do + case get_req_header(conn, "authorization") do + ["Bearer " <> token] -> String.trim(token) + ["bearer " <> token] -> String.trim(token) + _ -> nil + end + end + + defp first_header(conn, name) do + case get_req_header(conn, name) do + [value | _] -> value + _ -> nil + end + end + + defp connector_urls(conn) do + configured = + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_connector_urls, []) + + case configured do + urls when is_list(urls) and urls != [] -> + urls + + _ -> + scheme = if conn.scheme == :https, do: "wss", else: "ws" + host = conn.host + port = conn.port + + base = + cond do + scheme == "wss" and port in [443, nil] -> "#{scheme}://#{host}" + scheme == "ws" and port in [80, nil] -> "#{scheme}://#{host}" + true -> "#{scheme}://#{host}:#{port}" + end + + ["#{base}/_edge/tunnel/v1/ws"] + end + end + + defp tunnel_config(key, default) do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(key, default) + end + + defp unauthorized(conn, code) do + conn + |> put_status(:unauthorized) + |> json(%{error: code}) + end + + defp disabled(conn) do + conn + |> put_status(:not_found) + |> json(%{error: "tunnel_disabled"}) + end + + defp error(conn, status, code) do + conn + |> put_status(status) + |> json(%{error: code}) + end +end diff --git a/apps/elektrine_web/lib/elektrine_web/edge/connector.ex b/apps/elektrine_web/lib/elektrine_web/edge/connector.ex new file mode 100644 index 0000000..78dcbec --- /dev/null +++ b/apps/elektrine_web/lib/elektrine_web/edge/connector.ex @@ -0,0 +1,394 @@ +defmodule ElektrineWeb.Edge.Connector do + @moduledoc """ + Edge tunnel connector: holds authenticated WSS agent sessions and dispatches + HTTP requests from `DNSEdgeProxy` over multiplexed binary frames. + + Must run on a BEAM node with the Phoenix web endpoint enabled (KD-5c). + Feature-gated by `DNS_TUNNEL_ENABLED` / `:dns, :tunnel_enabled`. + """ + + use GenServer + + require Logger + + alias ElektrineWeb.Edge.TunnelFrame + + @tickets_table __MODULE__.Tickets + @max_response_body_bytes 50 * 1024 * 1024 + @default_dispatch_timeout_ms 30_000 + + # --- Public API ------------------------------------------------------------ + + def start_link(opts \\ []) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def child_spec(opts) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [opts]}, + type: :worker, + restart: :permanent + } + end + + def enabled? do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_enabled, false) + end + + @doc """ + DNSEdgeProxy dispatcher entry point. Returns `{:ok, conn}` or `{:error, reason}`. + """ + def dispatch(%Plug.Conn{} = conn, origin) do + if enabled?(), do: do_dispatch(conn, origin), else: {:error, :tunnel_disabled} + end + + def mint_session_ticket(%{id: _, zone_id: _} = tunnel) do + GenServer.call(__MODULE__, {:mint_ticket, tunnel}) + end + + def revoke_session_ticket(ticket) when is_binary(ticket) do + GenServer.call(__MODULE__, {:revoke_ticket, ticket}) + end + + def authenticate_ticket(ticket) when is_binary(ticket) do + GenServer.call(__MODULE__, {:authenticate_ticket, ticket}) + end + + def register_ws(tunnel_id, ws_pid, meta \\ %{}) + when is_binary(tunnel_id) and is_pid(ws_pid) do + GenServer.call(__MODULE__, {:register_ws, tunnel_id, ws_pid, meta}) + end + + def unregister_ws(tunnel_id, ws_pid) when is_binary(tunnel_id) and is_pid(ws_pid) do + GenServer.cast(__MODULE__, {:unregister_ws, tunnel_id, ws_pid}) + end + + def disconnect_tunnel(tunnel_id) when is_binary(tunnel_id) do + GenServer.call(__MODULE__, {:disconnect_tunnel, tunnel_id}) + end + + def session_online?(tunnel_id) when is_binary(tunnel_id) do + GenServer.call(__MODULE__, {:session_online?, tunnel_id}) + catch + :exit, _ -> false + end + + def max_response_body_bytes, do: @max_response_body_bytes + + def encode_frame(name, payload), do: TunnelFrame.encode(name, payload) + + defp session_ttl_seconds do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_session_ttl_seconds, 3_600) + end + + # --- GenServer ------------------------------------------------------------- + + @impl true + def init(_opts) do + Process.flag(:trap_exit, true) + + case :ets.whereis(@tickets_table) do + :undefined -> + :ets.new(@tickets_table, [ + :named_table, + :public, + :set, + read_concurrency: true, + write_concurrency: true + ]) + + _tid -> + :ok + end + + # sessions: tunnel_id => %{ws_pid, monitor_ref, meta} + # stream_seq: next odd stream id from edge + {:ok, %{sessions: %{}, stream_seq: 1}} + end + + @impl true + def handle_call({:mint_ticket, tunnel}, _from, state) do + ticket = "ets_" <> Base.url_encode64(:crypto.strong_rand_bytes(32), padding: false) + hash = ticket_hash(ticket) + expires_at = System.system_time(:second) + session_ttl_seconds() + allowed = Map.get(tunnel, :allowed_hosts) || [] + + :ets.insert(@tickets_table, { + hash, + %{ + tunnel_id: tunnel.id, + zone_id: tunnel.zone_id, + allowed_hosts: allowed, + expires_at: expires_at + } + }) + + {:reply, {:ok, ticket, expires_at}, state} + end + + def handle_call({:revoke_ticket, ticket}, _from, state) do + :ets.delete(@tickets_table, ticket_hash(ticket)) + {:reply, :ok, state} + end + + def handle_call({:authenticate_ticket, ticket}, _from, state) do + now = System.system_time(:second) + + reply = + case :ets.lookup(@tickets_table, ticket_hash(ticket)) do + [{_hash, %{expires_at: exp} = meta}] when exp > now -> + {:ok, meta} + + [{_hash, _meta}] -> + :ets.delete(@tickets_table, ticket_hash(ticket)) + {:error, :expired} + + [] -> + {:error, :invalid_ticket} + end + + {:reply, reply, state} + end + + def handle_call({:register_ws, tunnel_id, ws_pid, meta}, _from, state) do + # Drop prior session for this tunnel (single agent connection). + state = + case Map.get(state.sessions, tunnel_id) do + %{ws_pid: old_pid, monitor_ref: old_ref} when old_pid != ws_pid -> + Process.demonitor(old_ref, [:flush]) + send(old_pid, :tunnel_replaced) + %{state | sessions: Map.delete(state.sessions, tunnel_id)} + + _ -> + state + end + + ref = Process.monitor(ws_pid) + + sessions = + Map.put(state.sessions, tunnel_id, %{ + ws_pid: ws_pid, + monitor_ref: ref, + meta: meta + }) + + {:reply, :ok, %{state | sessions: sessions}} + end + + def handle_call({:disconnect_tunnel, tunnel_id}, _from, state) do + state = + case Map.pop(state.sessions, tunnel_id) do + {%{ws_pid: pid, monitor_ref: ref}, sessions} -> + Process.demonitor(ref, [:flush]) + send(pid, :tunnel_disconnect) + drop_tickets_for_tunnel(tunnel_id) + %{state | sessions: sessions} + + {nil, _} -> + drop_tickets_for_tunnel(tunnel_id) + state + end + + {:reply, :ok, state} + end + + def handle_call({:session_online?, tunnel_id}, _from, state) do + {:reply, Map.has_key?(state.sessions, tunnel_id), state} + end + + def handle_call({:open_stream, tunnel_id, request, waiter}, _from, state) do + case Map.get(state.sessions, tunnel_id) do + %{ws_pid: ws_pid} -> + stream_id = state.stream_seq + # Odd stream ids from edge (v1) + next_seq = stream_id + 2 + ref = make_ref() + send(ws_pid, {:open_stream, stream_id, request, waiter, ref}) + {:reply, {:ok, stream_id, ref}, %{state | stream_seq: next_seq}} + + nil -> + {:reply, {:error, :agent_down}, state} + end + end + + @impl true + def handle_cast({:unregister_ws, tunnel_id, ws_pid}, state) do + state = + case Map.get(state.sessions, tunnel_id) do + %{ws_pid: ^ws_pid, monitor_ref: ref} -> + Process.demonitor(ref, [:flush]) + %{state | sessions: Map.delete(state.sessions, tunnel_id)} + + _ -> + state + end + + {:noreply, state} + end + + @impl true + def handle_info({:DOWN, ref, :process, pid, _reason}, state) do + sessions = + Enum.reduce(state.sessions, %{}, fn + {tunnel_id, %{ws_pid: ^pid, monitor_ref: ^ref}}, acc -> + Logger.info("tunnel agent disconnected tunnel_id=#{tunnel_id}") + acc + + {tunnel_id, session}, acc -> + Map.put(acc, tunnel_id, session) + end) + + {:noreply, %{state | sessions: sessions}} + end + + def handle_info(_msg, state), do: {:noreply, state} + + # --- Dispatch -------------------------------------------------------------- + + defp do_dispatch(conn, origin) do + tunnel_id = origin_tunnel_id(origin) + host = origin_host(origin, conn) + + cond do + not is_binary(tunnel_id) or tunnel_id == "" -> + {:error, :missing_tunnel_id} + + not session_online?(tunnel_id) -> + {:error, :agent_down} + + true -> + case read_body_limited(conn) do + {:ok, body, conn} -> + request = %{ + method: conn.method, + path: request_path(conn), + headers: request_headers(conn, origin), + body: body, + host: host + } + + case proxy_via_agent(tunnel_id, request) do + {:ok, %{status: status, headers: headers, body: resp_body}} -> + conn = + headers + |> reject_hop_by_hop_headers() + |> Enum.reduce(conn, fn {name, value}, acc -> + Plug.Conn.put_resp_header( + acc, + String.downcase(to_string(name)), + to_string(value) + ) + end) + + conn = + conn + |> Plug.Conn.send_resp(status, resp_body) + |> Plug.Conn.halt() + + {:ok, conn} + + {:error, reason} -> + {:error, reason} + end + + {:too_large, _conn} -> + {:error, :request_too_large} + + {:error, _} -> + {:error, :bad_request} + end + end + end + + defp proxy_via_agent(tunnel_id, request) do + timeout = dispatch_timeout_ms() + waiter = self() + + case GenServer.call(__MODULE__, {:open_stream, tunnel_id, request, waiter}, 5_000) do + {:ok, stream_id, ref} -> + receive do + {:stream_complete, ^ref, ^stream_id, response} -> + {:ok, response} + + {:stream_error, ^ref, ^stream_id, reason} -> + {:error, reason} + after + timeout -> + {:error, :stream_timeout} + end + + {:error, reason} -> + {:error, reason} + end + catch + :exit, {:noproc, _} -> {:error, :connector_down} + :exit, {:timeout, _} -> {:error, :connector_timeout} + end + + defp read_body_limited(conn) do + max = 25 * 1024 * 1024 + + case Plug.Conn.read_body(conn, length: max, read_length: max) do + {:ok, body, conn} -> {:ok, body, conn} + {:more, _, conn} -> {:too_large, conn} + {:error, reason} -> {:error, reason} + end + end + + defp request_path(conn) do + query = if conn.query_string in [nil, ""], do: "", else: "?" <> conn.query_string + conn.request_path <> query + end + + @hop_by_hop ~w( + connection keep-alive proxy-authenticate proxy-authorization + te trailer transfer-encoding upgrade + ) + + defp request_headers(conn, origin) do + host_header = origin[:origin_host_header] || origin["origin_host_header"] || conn.host + + conn.req_headers + |> Enum.reject(fn {name, _} -> + down = String.downcase(name) + down in @hop_by_hop or down == "host" + end) + |> Kernel.++([{"host", host_header}]) + end + + defp reject_hop_by_hop_headers(headers) when is_list(headers) do + Enum.reject(headers, fn + {name, _} -> String.downcase(to_string(name)) in @hop_by_hop + _ -> false + end) + end + + defp reject_hop_by_hop_headers(_), do: [] + + defp origin_tunnel_id(%{tunnel_id: id}) when is_binary(id), do: id + defp origin_tunnel_id(%{"tunnel_id" => id}) when is_binary(id), do: id + defp origin_tunnel_id(_), do: nil + + defp origin_host(origin, conn), do: origin[:host] || origin["host"] || conn.host + + defp ticket_hash(ticket), do: :crypto.hash(:sha256, ticket) |> Base.encode16(case: :lower) + + defp dispatch_timeout_ms do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(:tunnel_dispatch_timeout_ms, @default_dispatch_timeout_ms) + end + + defp drop_tickets_for_tunnel(tunnel_id) do + :ets.foldl( + fn {hash, meta}, _acc -> + if meta.tunnel_id == tunnel_id, do: :ets.delete(@tickets_table, hash) + nil + end, + nil, + @tickets_table + ) + end +end diff --git a/apps/elektrine_web/lib/elektrine_web/edge/tunnel_frame.ex b/apps/elektrine_web/lib/elektrine_web/edge/tunnel_frame.ex new file mode 100644 index 0000000..235fa62 --- /dev/null +++ b/apps/elektrine_web/lib/elektrine_web/edge/tunnel_frame.ex @@ -0,0 +1,209 @@ +defmodule ElektrineWeb.Edge.TunnelFrame do + @moduledoc """ + Length-prefixed binary frame codec for the edge tunnel WSS protocol (v1). + + Wire format: + + <> + + Max payload size is 1 MiB. Structured control frames use JSON payloads; + `STREAM_BODY` carries a binary header (`stream_id` + flags) plus raw bytes. + + Lives in `elektrine_web` so the connector and agent share one codec without + a compile cycle (dns depends on web). + """ + + @max_payload_bytes 1 * 1024 * 1024 + + @type_auth 0x01 + @type_auth_ok 0x02 + @type_auth_fail 0x03 + @type_stream_open 0x10 + @type_stream_headers 0x11 + @type_stream_body 0x12 + @type_stream_rst 0x13 + @type_ping 0x20 + @type_pong 0x21 + + @type_names %{ + @type_auth => :auth, + @type_auth_ok => :auth_ok, + @type_auth_fail => :auth_fail, + @type_stream_open => :stream_open, + @type_stream_headers => :stream_headers, + @type_stream_body => :stream_body, + @type_stream_rst => :stream_rst, + @type_ping => :ping, + @type_pong => :pong + } + + @name_types Map.new(@type_names, fn {k, v} -> {v, k} end) + + @fin_flag 0x01 + + @type frame :: + {:auth, map()} + | {:auth_ok, map()} + | {:auth_fail, map()} + | {:stream_open, map()} + | {:stream_headers, map()} + | {:stream_body, %{stream_id: non_neg_integer(), fin: boolean(), data: binary()}} + | {:stream_rst, map()} + | {:ping, binary()} + | {:pong, binary()} + + def max_payload_bytes, do: @max_payload_bytes + def fin_flag, do: @fin_flag + + # STREAM_BODY payload is <> + def stream_body_max_data_bytes, do: @max_payload_bytes - 5 + + def type_byte(name) when is_atom(name), do: Map.fetch!(@name_types, name) + def type_name(byte) when is_integer(byte), do: Map.get(@type_names, byte) + + @doc """ + Encode a named frame to wire bytes. + """ + def encode(name, payload) when is_atom(name) do + type = type_byte(name) + body = encode_payload(name, payload) + + if byte_size(body) > @max_payload_bytes do + {:error, :payload_too_large} + else + {:ok, <>} + end + end + + @doc """ + Encode and raise on error (tests / internal call sites). + """ + def encode!(name, payload) do + case encode(name, payload) do + {:ok, bin} -> bin + {:error, reason} -> raise ArgumentError, "frame encode failed: #{inspect(reason)}" + end + end + + @doc """ + Decode one complete frame from a buffer. + + Returns `{:ok, frame, rest}`, `{:incomplete, buffer}`, or `{:error, reason}`. + """ + def decode(buffer) when is_binary(buffer) and byte_size(buffer) < 5 do + {:incomplete, buffer} + end + + def decode(<>) + when payload_len > @max_payload_bytes do + {:error, :payload_too_large} + end + + def decode(<> = buffer) + when is_integer(payload_len) do + if byte_size(rest) < payload_len do + {:incomplete, buffer} + else + payload = binary_part(rest, 0, payload_len) + remaining = binary_part(rest, payload_len, byte_size(rest) - payload_len) + + case decode_payload(type, payload) do + {:ok, frame} -> {:ok, frame, remaining} + {:error, reason} -> {:error, reason} + end + end + end + + def decode(buffer) when is_binary(buffer) do + {:incomplete, buffer} + end + + def decode(_), do: {:error, :invalid_frame} + + @doc """ + Decode all complete frames from a buffer. Returns `{frames, remainder}`. + """ + def decode_all(buffer) when is_binary(buffer) do + do_decode_all(buffer, []) + end + + defp do_decode_all(buffer, acc) do + case decode(buffer) do + {:ok, frame, rest} -> do_decode_all(rest, [frame | acc]) + {:incomplete, rest} -> {Enum.reverse(acc), rest} + {:error, reason} -> {:error, reason, Enum.reverse(acc), buffer} + end + end + + defp encode_payload(:stream_body, %{stream_id: stream_id, data: data} = payload) + when is_integer(stream_id) and is_binary(data) do + flags = if Map.get(payload, :fin, false), do: @fin_flag, else: 0 + <> + end + + defp encode_payload(:ping, opaque) when is_binary(opaque) and byte_size(opaque) == 8, do: opaque + defp encode_payload(:pong, opaque) when is_binary(opaque) and byte_size(opaque) == 8, do: opaque + + defp encode_payload(:ping, _), do: :crypto.strong_rand_bytes(8) + defp encode_payload(:pong, _), do: :crypto.strong_rand_bytes(8) + + defp encode_payload(_name, payload) when is_map(payload) do + Jason.encode!(stringify_keys(payload)) + end + + defp decode_payload(@type_stream_body, <>) do + {:ok, + {:stream_body, %{stream_id: stream_id, fin: band(flags, @fin_flag) == @fin_flag, data: data}}} + end + + defp decode_payload(@type_stream_body, _), do: {:error, :invalid_stream_body} + + defp decode_payload(@type_ping, opaque) when byte_size(opaque) == 8, do: {:ok, {:ping, opaque}} + defp decode_payload(@type_pong, opaque) when byte_size(opaque) == 8, do: {:ok, {:pong, opaque}} + defp decode_payload(@type_ping, _), do: {:error, :invalid_ping} + defp decode_payload(@type_pong, _), do: {:error, :invalid_pong} + + defp decode_payload(type, payload) when is_map_key(@type_names, type) do + name = Map.fetch!(@type_names, type) + + case Jason.decode(payload) do + {:ok, map} when is_map(map) -> {:ok, {name, atomize_known_keys(map)}} + {:ok, _} -> {:error, :invalid_json_payload} + {:error, _} -> {:error, :invalid_json_payload} + end + end + + defp decode_payload(_, _), do: {:error, :unknown_frame_type} + + defp stringify_keys(map) do + Map.new(map, fn + {k, v} when is_atom(k) -> {Atom.to_string(k), stringify_value(v)} + {k, v} -> {k, stringify_value(v)} + end) + end + + defp stringify_value(list) when is_list(list), do: Enum.map(list, &stringify_value/1) + defp stringify_value(map) when is_map(map), do: stringify_keys(map) + defp stringify_value(other), do: other + + defp atomize_known_keys(map) do + Map.new(map, fn + {"stream_id", v} when is_integer(v) -> {:stream_id, v} + {"status", v} when is_integer(v) -> {:status, v} + {"method", v} when is_binary(v) -> {:method, v} + {"path", v} when is_binary(v) -> {:path, v} + {"headers", v} when is_list(v) -> {:headers, v} + {"session_ticket", v} when is_binary(v) -> {:session_ticket, v} + {"agent_version", v} when is_binary(v) -> {:agent_version, v} + {"max_streams", v} when is_integer(v) -> {:max_streams, v} + {"idle_timeout_ms", v} when is_integer(v) -> {:idle_timeout_ms, v} + {"server_time", v} -> {:server_time, v} + {"reason", v} when is_binary(v) -> {:reason, v} + {"error", v} when is_binary(v) -> {:error, v} + {"error_code", v} -> {:error_code, v} + {k, v} -> {k, v} + end) + end + + defp band(a, b), do: Bitwise.band(a, b) +end diff --git a/apps/elektrine_web/lib/elektrine_web/edge/tunnel_websock.ex b/apps/elektrine_web/lib/elektrine_web/edge/tunnel_websock.ex new file mode 100644 index 0000000..d818e3b --- /dev/null +++ b/apps/elektrine_web/lib/elektrine_web/edge/tunnel_websock.ex @@ -0,0 +1,442 @@ +defmodule ElektrineWeb.Edge.TunnelWebSock do + @moduledoc """ + WebSock handler for `/_edge/tunnel/v1/ws`. + + Agents authenticate with a session ticket (prefer first AUTH frame; query + param still accepted for compatibility) then multiplex HTTP streams using + `ElektrineWeb.Edge.TunnelFrame`. + """ + + @behaviour WebSock + + require Logger + + alias ElektrineWeb.Edge.Connector + alias ElektrineWeb.Edge.TunnelFrame + + @type state :: %{ + optional(:tunnel_id) => String.t(), + optional(:allowed_hosts) => [String.t()], + optional(:zone_id) => integer(), + optional(:authenticated) => boolean(), + optional(:closing) => boolean(), + buffer: binary(), + streams: map(), + max_streams: pos_integer(), + idle_timeout_ms: pos_integer() + } + + @impl true + def init(opts) do + opts = Map.new(opts) + + state = %{ + buffer: "", + streams: %{}, + max_streams: tunnel_config(:tunnel_max_streams, 16), + idle_timeout_ms: tunnel_config(:tunnel_idle_timeout_ms, 60_000), + authenticated: false + } + + case Map.get(opts, :session_ticket) do + ticket when is_binary(ticket) and ticket != "" -> + case authenticate(ticket, state) do + {:ok, state} -> + {:push, auth_ok_frame(state), state} + + {:error, reason} -> + frame = TunnelFrame.encode!(:auth_fail, %{reason: to_string(reason)}) + # Close promptly on failed auth (no idle hold-open). + {:stop, :normal, [{:binary, frame}], Map.put(state, :closing, true)} + end + + _ -> + {:ok, state} + end + end + + @impl true + def handle_in({data, opcode: :binary}, state) when is_binary(data) do + if Map.get(state, :closing) do + {:ok, state} + else + buffer = state.buffer <> data + process_buffer(buffer, state, []) + end + end + + def handle_in({_data, opcode: :text}, state) do + # Text frames are not part of the v1 protocol. + {:ok, state} + end + + def handle_in(_other, state), do: {:ok, state} + + @impl true + def handle_info({:open_stream, stream_id, request, waiter, ref}, state) do + cond do + state.authenticated != true -> + send(waiter, {:stream_error, ref, stream_id, :not_authenticated}) + {:ok, state} + + map_size(state.streams) >= state.max_streams -> + send(waiter, {:stream_error, ref, stream_id, :max_streams}) + {:ok, state} + + not host_allowed?(state, Map.get(request, :host)) -> + send(waiter, {:stream_error, ref, stream_id, :host_not_allowed}) + {:ok, state} + + true -> + body = Map.get(request, :body, "") || "" + + case encode_open_and_body(stream_id, request, body) do + {:ok, frames} -> + streams = + Map.put(state.streams, stream_id, %{ + stream_id: stream_id, + waiter: waiter, + ref: ref, + status: nil, + headers: [], + body: <<>>, + body_bytes: 0 + }) + + {:push, Enum.map(frames, &{:binary, &1}), %{state | streams: streams}} + + {:error, reason} -> + send(waiter, {:stream_error, ref, stream_id, reason}) + {:ok, state} + end + end + end + + def handle_info(:tunnel_disconnect, state) do + {:stop, :normal, state} + end + + def handle_info(:tunnel_replaced, state) do + {:stop, :normal, state} + end + + def handle_info(_msg, state), do: {:ok, state} + + @impl true + def terminate(_reason, state) do + if tunnel_id = state[:tunnel_id] do + Connector.unregister_ws(tunnel_id, self()) + end + + for {stream_id, %{waiter: waiter, ref: ref}} <- state.streams do + if is_pid(waiter) do + send(waiter, {:stream_error, ref, stream_id, :agent_down}) + end + end + + :ok + end + + # --- Internals ------------------------------------------------------------- + + defp encode_open_and_body(stream_id, request, body) do + payload = %{ + stream_id: stream_id, + method: Map.get(request, :method, "GET"), + path: Map.get(request, :path, "/"), + headers: encode_headers(Map.get(request, :headers, [])), + has_body: byte_size(body) > 0 + } + + with {:ok, open} <- TunnelFrame.encode(:stream_open, payload), + {:ok, body_frames} <- encode_body_chunks(stream_id, body) do + {:ok, [open | body_frames]} + else + {:error, reason} -> {:error, reason} + end + end + + defp encode_body_chunks(_stream_id, ""), do: {:ok, []} + + defp encode_body_chunks(stream_id, body) when is_binary(body) do + max = TunnelFrame.stream_body_max_data_bytes() + chunks = chunk_binary(body, max) + last = length(chunks) - 1 + + chunks + |> Enum.with_index() + |> Enum.reduce_while({:ok, []}, fn {chunk, idx}, {:ok, acc} -> + case TunnelFrame.encode(:stream_body, %{ + stream_id: stream_id, + fin: idx == last, + data: chunk + }) do + {:ok, frame} -> {:cont, {:ok, acc ++ [frame]}} + {:error, reason} -> {:halt, {:error, reason}} + end + end) + end + + defp chunk_binary(bin, size) when is_binary(bin) and size > 0 do + do_chunk(bin, size, []) + end + + defp do_chunk(<<>>, _size, acc), do: Enum.reverse(acc) + + defp do_chunk(bin, size, acc) when byte_size(bin) <= size do + Enum.reverse([bin | acc]) + end + + defp do_chunk(bin, size, acc) do + chunk = binary_part(bin, 0, size) + rest = binary_part(bin, size, byte_size(bin) - size) + do_chunk(rest, size, [chunk | acc]) + end + + defp process_buffer(buffer, state, pushes) do + case TunnelFrame.decode(buffer) do + {:ok, frame, rest} -> + case handle_frame(frame, state) do + {:ok, state, out} -> + process_buffer(rest, state, pushes ++ out) + + {:stop, state, out} -> + {:stop, :normal, pushes ++ out, Map.put(state, :closing, true)} + end + + {:incomplete, rest} -> + reply_pushes(pushes, %{state | buffer: rest}) + + {:error, reason} -> + Logger.warning("tunnel frame decode error: #{inspect(reason)}") + {:stop, :normal, %{state | buffer: ""}} + end + end + + defp reply_pushes([], state), do: {:ok, state} + defp reply_pushes(pushes, state), do: {:push, pushes, state} + + defp handle_frame({:auth, payload}, state) do + ticket = payload[:session_ticket] || payload["session_ticket"] + + case authenticate(ticket, state) do + {:ok, state} -> + {:ok, state, auth_ok_frame(state)} + + {:error, reason} -> + frame = {:binary, TunnelFrame.encode!(:auth_fail, %{reason: to_string(reason)})} + {:stop, state, [frame]} + end + end + + defp handle_frame({:ping, opaque}, state) do + {:ok, state, [{:binary, TunnelFrame.encode!(:pong, opaque)}]} + end + + defp handle_frame({:pong, _opaque}, state), do: {:ok, state, []} + + defp handle_frame({:stream_headers, payload}, state) do + stream_id = payload[:stream_id] || payload["stream_id"] + + case Map.get(state.streams, stream_id) do + nil -> + {:ok, state, []} + + stream -> + status = payload[:status] || payload["status"] || 502 + headers = normalize_headers(payload[:headers] || payload["headers"] || []) + stream = %{stream | status: status, headers: headers} + {:ok, %{state | streams: Map.put(state.streams, stream_id, stream)}, []} + end + end + + defp handle_frame({:stream_body, %{stream_id: stream_id, fin: fin, data: data}}, state) do + case Map.get(state.streams, stream_id) do + nil -> + {:ok, state, []} + + stream -> + new_size = stream.body_bytes + byte_size(data) + + if new_size > Connector.max_response_body_bytes() do + send(stream.waiter, {:stream_error, stream.ref, stream_id, :response_too_large}) + + rst = + TunnelFrame.encode!(:stream_rst, %{stream_id: stream_id, error: "response_too_large"}) + + {:ok, %{state | streams: Map.delete(state.streams, stream_id)}, [{:binary, rst}]} + else + stream = %{ + stream + | body: stream.body <> data, + body_bytes: new_size + } + + if fin do + complete_stream(state, stream_id, stream) + else + {:ok, %{state | streams: Map.put(state.streams, stream_id, stream)}, []} + end + end + end + end + + defp handle_frame({:stream_rst, payload}, state) do + stream_id = payload[:stream_id] || payload["stream_id"] + + case Map.pop(state.streams, stream_id) do + {nil, _} -> + {:ok, state, []} + + {%{waiter: waiter, ref: ref}, streams} -> + error = payload[:error] || payload["error"] || "rst" + send(waiter, {:stream_error, ref, stream_id, error}) + {:ok, %{state | streams: streams}, []} + end + end + + defp handle_frame(_other, state), do: {:ok, state, []} + + defp complete_stream(state, stream_id, stream) do + status = stream.status || 502 + + response = %{ + status: status, + headers: stream.headers, + body: stream.body + } + + send(stream.waiter, {:stream_complete, stream.ref, stream_id, response}) + {:ok, %{state | streams: Map.delete(state.streams, stream_id)}, []} + end + + defp authenticate(ticket, state) when is_binary(ticket) do + case Connector.authenticate_ticket(ticket) do + {:ok, meta} -> + tunnel_id = meta.tunnel_id + :ok = Connector.register_ws(tunnel_id, self(), meta) + + touch_tunnel_last_seen(tunnel_id) + + {:ok, + Map.merge(state, %{ + authenticated: true, + tunnel_id: tunnel_id, + zone_id: meta.zone_id, + allowed_hosts: meta.allowed_hosts || [] + })} + + {:error, reason} -> + {:error, reason} + end + end + + defp authenticate(_, _), do: {:error, :missing_ticket} + + defp auth_ok_frame(state) do + frame = + TunnelFrame.encode!(:auth_ok, %{ + idle_timeout_ms: state.idle_timeout_ms, + server_time: DateTime.utc_now() |> DateTime.to_iso8601() + }) + + [{:binary, frame}] + end + + defp touch_tunnel_last_seen(tunnel_id) do + tunnels = tunnels_mod() + + if function_exported?(tunnels, :get_tunnel, 1) and + function_exported?(tunnels, :touch_last_seen, 1) do + case tunnels.get_tunnel(tunnel_id) do + %{__struct__: _} = tunnel -> tunnels.touch_last_seen(tunnel) + _ -> :ok + end + else + :ok + end + end + + defp tunnels_mod, do: Elektrine.DNS.Tunnels + + defp tunnel_config(key, default) do + Application.get_env(:elektrine, :dns, []) + |> Keyword.get(key, default) + end + + defp host_allowed?(%{allowed_hosts: hosts}, host) when is_list(hosts) do + host_allowed_list?(hosts, host || "") + end + + defp host_allowed?(_, _), do: true + + # Prefer Tunnel.host_matches?/2 when dns is loaded; local fallback for tests. + defp host_allowed_list?(hosts, host) when is_list(hosts) and is_binary(host) do + tunnel = Elektrine.DNS.Tunnel + + if function_exported?(tunnel, :host_matches?, 2) do + tunnel.host_matches?(hosts, host) + else + local_host_matches?(hosts, host) + end + end + + defp host_allowed_list?(_, _), do: false + + defp local_host_matches?([], _host), do: true + + defp local_host_matches?(hosts, host) do + normalized = normalize_host(host) + + Enum.any?(hosts, fn allowed -> + entry = normalize_host(to_string(allowed)) + + cond do + entry == nil or entry == "" -> + false + + entry == normalized -> + true + + String.starts_with?(entry, "*.") -> + rest = String.trim_leading(entry, "*.") + + case String.split(normalized, ".", parts: 2) do + [_label, ^rest] -> true + _ -> false + end + + true -> + false + end + end) + end + + defp normalize_host(host) when is_binary(host) do + host + |> String.trim() + |> String.trim_trailing(".") + |> String.downcase() + end + + defp normalize_host(_), do: nil + + defp encode_headers(headers) do + Enum.map(headers, fn + {k, v} -> [to_string(k), to_string(v)] + [k, v] -> [to_string(k), to_string(v)] + other -> other + end) + end + + defp normalize_headers(headers) when is_list(headers) do + Enum.flat_map(headers, fn + [k, v] when is_binary(k) -> [{k, to_string(v)}] + {k, v} -> [{to_string(k), to_string(v)}] + %{"name" => k, "value" => v} -> [{to_string(k), to_string(v)}] + _ -> [] + end) + end + + defp normalize_headers(_), do: [] +end diff --git a/apps/elektrine_web/lib/elektrine_web/plugs/dns_edge_proxy.ex b/apps/elektrine_web/lib/elektrine_web/plugs/dns_edge_proxy.ex index 030bf14..392de0b 100644 --- a/apps/elektrine_web/lib/elektrine_web/plugs/dns_edge_proxy.ex +++ b/apps/elektrine_web/lib/elektrine_web/plugs/dns_edge_proxy.ex @@ -249,15 +249,17 @@ defmodule ElektrineWeb.Plugs.DNSEdgeProxy do end # Tunnel origins never dial the control-plane HTTP client (SSRF boundary). - # The Edge.Connector session path lands in a later PR; until then, fail closed. + # Live sessions are held by ElektrineWeb.Edge.Connector over outbound WSS. defp proxy_tunnel(conn, origin) do case tunnel_dispatcher().(conn, origin) do {:ok, %Plug.Conn{} = conn} -> if conn.halted, do: conn, else: halt(conn) - {:error, _reason} -> + {:error, reason} -> + {status, body} = tunnel_error_response(reason) + conn - |> send_resp(502, "Bad Gateway") + |> send_resp(status, body) |> halt() _other -> @@ -268,6 +270,15 @@ defmodule ElektrineWeb.Plugs.DNSEdgeProxy do end defp proxy_public(conn, origin, rewrites) do +defp tunnel_error_response(:request_too_large), do: {413, "Payload Too Large"} + defp tunnel_error_response(:bad_request), do: {400, "Bad Request"} + defp tunnel_error_response(:host_not_allowed), do: {403, "Forbidden"} + defp tunnel_error_response(:max_streams), do: {503, "Service Unavailable"} + defp tunnel_error_response(:tunnel_disabled), do: {404, "Not Found"} + defp tunnel_error_response(:missing_tunnel_id), do: {502, "Bad Gateway"} + defp tunnel_error_response(_), do: {502, "Bad Gateway"} + + defp proxy_public(conn, origin) do case read_proxy_body(conn) do {:ok, body, conn} -> request = @@ -434,7 +445,14 @@ defmodule ElektrineWeb.Plugs.DNSEdgeProxy do end end - defp default_tunnel_dispatcher(_conn, _origin), do: {:error, :tunnel_unavailable} + defp default_tunnel_dispatcher(conn, origin) do + if Code.ensure_loaded?(ElektrineWeb.Edge.Connector) and + function_exported?(ElektrineWeb.Edge.Connector, :dispatch, 2) do + ElektrineWeb.Edge.Connector.dispatch(conn, origin) + else + {:error, :tunnel_unavailable} + end + end defp http_opts do [ diff --git a/apps/elektrine_web/lib/elektrine_web/router.ex b/apps/elektrine_web/lib/elektrine_web/router.ex index 198ea57..74e6500 100644 --- a/apps/elektrine_web/lib/elektrine_web/router.ex +++ b/apps/elektrine_web/lib/elektrine_web/router.ex @@ -277,6 +277,14 @@ defmodule ElektrineWeb.Router do plug(ElektrineWeb.Plugs.PATAuth, scopes: ["write:dns"]) end + pipeline :api_pat_edge_read_scope do + plug(ElektrineWeb.Plugs.PATAuth, scopes: ["read:edge", "write:edge"], any: true) + end + + pipeline :api_pat_edge_write_scope do + plug(ElektrineWeb.Plugs.PATAuth, scopes: ["write:edge"]) + end + pipeline :api_pat_nerve_read_scope do plug(ElektrineWeb.Plugs.PATAuth, scopes: ["read:nerve", "write:nerve"], @@ -437,6 +445,15 @@ defmodule ElektrineWeb.Router do get("/callback", EdgeAccessController, :callback) get("/complete", EdgeAccessController, :complete) +# Tunnel agent control plane + WSS (auth via etn_ tunnel token / session ticket). + # Gated by DNS_TUNNEL_ENABLED; returns 404 when disabled. + scope "/_edge/tunnel/v1", ElektrineWeb do + pipe_through([:api]) + + post("/register", TunnelController, :register) + post("/refresh", TunnelController, :refresh) + post("/disconnect", TunnelController, :disconnect) + get("/ws", TunnelController, :ws) end # Media proxy for federation privacy (no auth required) @@ -1696,6 +1713,18 @@ defmodule ElektrineWeb.Router do ElektrineWeb.Routes.DNS.api_write_routes() end + scope "/api/ext/v1/dns", ElektrineWeb.API do + pipe_through([:api_pat_authenticated, :api_pat_edge_read_scope]) + + ElektrineWeb.Routes.DNS.api_edge_read_routes() + end + + scope "/api/ext/v1/dns", ElektrineWeb.API do + pipe_through([:api_pat_authenticated, :api_pat_edge_write_scope]) + + ElektrineWeb.Routes.DNS.api_edge_write_routes() + end + scope "/api/ext/v1/nerve", ElektrineWeb.API do pipe_through([:api_nerve_authenticated, :api_pat_nerve_write_scope]) diff --git a/apps/elektrine_web/lib/elektrine_web/routes/dns.ex b/apps/elektrine_web/lib/elektrine_web/routes/dns.ex index 767ca7b..35dc88d 100644 --- a/apps/elektrine_web/lib/elektrine_web/routes/dns.ex +++ b/apps/elektrine_web/lib/elektrine_web/routes/dns.ex @@ -56,6 +56,25 @@ defmodule ElektrineWeb.Routes.DNS do end end + defmacro api_edge_read_routes do + quote do + scope "/", alias: false do + get("/zones/:zone_id/tunnels", ElektrineDNSWeb.API.DNSController, :list_tunnels) + get("/zones/:zone_id/tunnels/:id", ElektrineDNSWeb.API.DNSController, :show_tunnel) + end + end + end + + defmacro api_edge_write_routes do + quote do + scope "/", alias: false do + post("/zones/:zone_id/tunnels", ElektrineDNSWeb.API.DNSController, :create_tunnel) + put("/zones/:zone_id/tunnels/:id", ElektrineDNSWeb.API.DNSController, :update_tunnel) + delete("/zones/:zone_id/tunnels/:id", ElektrineDNSWeb.API.DNSController, :revoke_tunnel) + end + end + end + defmacro main_live_routes do quote do scope "/", alias: false do diff --git a/apps/elektrine_web/mix.exs b/apps/elektrine_web/mix.exs index 6aa4a93..c685c0b 100644 --- a/apps/elektrine_web/mix.exs +++ b/apps/elektrine_web/mix.exs @@ -69,6 +69,8 @@ defmodule ElektrineWeb.MixProject do ElektrineEmailWeb.DAV.AddressBookController, ElektrineDNSWeb.API.DNSController, ElektrineDNSWeb.DNSLive.Index, + Elektrine.DNS.Tunnels, + Elektrine.DNS.Tunnel, ElektrineSocialWeb.DiscussionsLive.Community, ElektrineSocialWeb.DiscussionsLive.Index, ElektrineSocialWeb.DiscussionsLive.Post, diff --git a/apps/elektrine_web/test/elektrine_web/controllers/api/dns_tunnels_controller_test.exs b/apps/elektrine_web/test/elektrine_web/controllers/api/dns_tunnels_controller_test.exs new file mode 100644 index 0000000..db3ee1f --- /dev/null +++ b/apps/elektrine_web/test/elektrine_web/controllers/api/dns_tunnels_controller_test.exs @@ -0,0 +1,86 @@ +defmodule ElektrineWeb.API.DNSTunnelsControllerTest do + use ElektrineWeb.ConnCase, async: false + + import Elektrine.AccountsFixtures + + alias Elektrine.Developer + alias Elektrine.DNS + + setup do + user = user_fixture() + + {:ok, zone} = + DNS.create_zone(user, %{"domain" => "api-tun-#{System.unique_integer([:positive])}.test"}) + + %{user: user, zone: zone} + end + + test "write:edge can mint and list tunnels; write:dns cannot mint", %{ + conn: conn, + user: user, + zone: zone + } do + # write:dns only is denied + deny = + conn + |> with_pat(user.id, ["write:dns"]) + |> post("/api/ext/v1/dns/zones/#{zone.id}/tunnels", %{ + "tunnel" => %{"name" => "denied"} + }) + + assert deny.status in [401, 403] + + # write:edge mints and returns token once + create = + conn + |> with_pat(user.id, ["write:edge"]) + |> post("/api/ext/v1/dns/zones/#{zone.id}/tunnels", %{ + "tunnel" => %{ + "name" => "home", + "allowed_hosts" => ["app.#{zone.domain}"] + } + }) + + assert %{"data" => %{"tunnel" => tunnel}} = json_response(create, 201) + assert tunnel["name"] == "home" + assert String.starts_with?(tunnel["token"], "etn_") + assert tunnel["allowed_hosts"] == ["app.#{zone.domain}"] + + # read:edge lists without token + list = + conn + |> with_pat(user.id, ["read:edge"]) + |> get("/api/ext/v1/dns/zones/#{zone.id}/tunnels") + + assert %{"data" => %{"tunnels" => tunnels}} = json_response(list, 200) + assert Enum.any?(tunnels, &(&1["id"] == tunnel["id"])) + refute Enum.any?(tunnels, &Map.has_key?(&1, "token")) + + # revoke with write:edge + revoke = + conn + |> with_pat(user.id, ["write:edge"]) + |> delete("/api/ext/v1/dns/zones/#{zone.id}/tunnels/#{tunnel["id"]}") + + assert json_response(revoke, 200)["data"]["message"] =~ "revoked" + end + + test "read:dns cannot list tunnels", %{conn: conn, user: user, zone: zone} do + conn = + conn + |> with_pat(user.id, ["read:dns"]) + |> get("/api/ext/v1/dns/zones/#{zone.id}/tunnels") + + assert conn.status in [401, 403] + end + + defp with_pat(conn, user_id, scopes) do + {:ok, token} = + Developer.create_api_token(user_id, %{ + name: "test-#{System.unique_integer([:positive])}", + scopes: scopes + }) + + put_req_header(conn, "authorization", "Bearer #{token.token}") + end +end diff --git a/apps/elektrine_web/test/elektrine_web/controllers/tunnel_controller_test.exs b/apps/elektrine_web/test/elektrine_web/controllers/tunnel_controller_test.exs new file mode 100644 index 0000000..50efc0f --- /dev/null +++ b/apps/elektrine_web/test/elektrine_web/controllers/tunnel_controller_test.exs @@ -0,0 +1,96 @@ +defmodule ElektrineWeb.TunnelControllerTest do + use ElektrineWeb.ConnCase, async: false + + import Elektrine.AccountsFixtures + + alias Elektrine.DNS + alias Elektrine.DNS.Tunnels + alias ElektrineWeb.Edge.Connector + + setup do + previous = Application.get_env(:elektrine, :dns, []) + + Application.put_env( + :elektrine, + :dns, + Keyword.merge(previous, tunnel_enabled: true) + ) + + on_exit(fn -> Application.put_env(:elektrine, :dns, previous) end) + + case Process.whereis(Connector) do + nil -> start_supervised!(Connector) + _ -> :ok + end + + user = user_fixture() + + {:ok, zone} = + DNS.create_zone(user, %{"domain" => "tc-#{System.unique_integer([:positive])}.test"}) + + {:ok, tunnel} = Tunnels.create_tunnel(zone, %{"name" => "lab"}, user) + + %{user: user, zone: zone, tunnel: tunnel, raw_token: tunnel.token} + end + + test "register exchanges etn token for session ticket", %{ + conn: conn, + tunnel: tunnel, + raw_token: token + } do + conn = + conn + |> put_req_header("authorization", "Bearer #{token}") + |> post("/_edge/tunnel/v1/register", %{}) + + assert %{ + "tunnel_id" => tunnel_id, + "session_ticket" => ticket, + "connector_urls" => urls, + "idle_timeout_ms" => idle + } = json_response(conn, 200) + + assert tunnel_id == tunnel.id + assert String.starts_with?(ticket, "ets_") + assert is_list(urls) and urls != [] + assert is_integer(idle) + end + + test "register rejects invalid token", %{conn: conn} do + conn = + conn + |> put_req_header("authorization", "Bearer etn_invalid") + |> post("/_edge/tunnel/v1/register", %{}) + + assert json_response(conn, 401)["error"] in ["invalid_token", "missing_token"] + end + + test "register returns 404 when tunnels disabled", %{conn: conn, raw_token: token} do + previous = Application.get_env(:elektrine, :dns, []) + Application.put_env(:elektrine, :dns, Keyword.put(previous, :tunnel_enabled, false)) + + conn = + conn + |> put_req_header("authorization", "Bearer #{token}") + |> post("/_edge/tunnel/v1/register", %{}) + + assert json_response(conn, 404)["error"] == "tunnel_disabled" + end + + test "refresh and disconnect", %{conn: conn, tunnel: tunnel, raw_token: token} do + refresh = + conn + |> put_req_header("authorization", "Bearer #{token}") + |> post("/_edge/tunnel/v1/refresh", %{}) + + assert %{"session_ticket" => ticket} = json_response(refresh, 200) + assert String.starts_with?(ticket, "ets_") + + disc = + conn + |> put_req_header("authorization", "Bearer #{token}") + |> post("/_edge/tunnel/v1/disconnect", %{}) + + assert json_response(disc, 200)["tunnel_id"] == tunnel.id + end +end diff --git a/apps/elektrine_web/test/elektrine_web/edge/connector_test.exs b/apps/elektrine_web/test/elektrine_web/edge/connector_test.exs new file mode 100644 index 0000000..cd6d0af --- /dev/null +++ b/apps/elektrine_web/test/elektrine_web/edge/connector_test.exs @@ -0,0 +1,305 @@ +defmodule ElektrineWeb.Edge.ConnectorTest do + use ElektrineWeb.ConnCase, async: false + + import Elektrine.AccountsFixtures + + alias Elektrine.DNS + alias Elektrine.DNS.Tunnels + alias ElektrineWeb.Edge.Connector + alias ElektrineWeb.Edge.TunnelFrame + alias ElektrineWeb.Edge.TunnelWebSock + + setup do + previous = Application.get_env(:elektrine, :dns, []) + + Application.put_env( + :elektrine, + :dns, + Keyword.merge(previous, tunnel_enabled: true) + ) + + on_exit(fn -> Application.put_env(:elektrine, :dns, previous) end) + + ensure_connector() + + user = user_fixture() + + {:ok, zone} = + DNS.create_zone(user, %{"domain" => "conn-#{System.unique_integer([:positive])}.test"}) + + {:ok, tunnel} = + Tunnels.create_tunnel(zone, %{"name" => "lab", "allowed_hosts" => []}, user) + + %{user: user, zone: zone, tunnel: tunnel} + end + + test "mint and authenticate session ticket", %{tunnel: tunnel} do + assert {:ok, ticket, _exp} = Connector.mint_session_ticket(tunnel) + assert String.starts_with?(ticket, "ets_") + assert {:ok, meta} = Connector.authenticate_ticket(ticket) + assert meta.tunnel_id == tunnel.id + end + + test "AUTH frame golden path via TunnelWebSock.init", %{tunnel: tunnel} do + assert {:ok, ticket, _} = Connector.mint_session_ticket(tunnel) + + assert {:push, frames, state} = TunnelWebSock.init(%{session_ticket: ticket}) + assert state.authenticated + assert state.tunnel_id == tunnel.id + assert Connector.session_online?(tunnel.id) + + assert Enum.any?(frames, fn + {:binary, bin} -> match?({:ok, {:auth_ok, _}, _}, TunnelFrame.decode(bin)) + _ -> false + end) + end + + test "AUTH frame via first AUTH message", %{tunnel: tunnel} do + assert {:ok, ticket, _} = Connector.mint_session_ticket(tunnel) + assert {:ok, state} = TunnelWebSock.init(%{}) + + auth = + TunnelFrame.encode!(:auth, %{session_ticket: ticket, agent_version: "test", max_streams: 8}) + + assert {:push, frames, state} = TunnelWebSock.handle_in({auth, opcode: :binary}, state) + assert state.authenticated + + assert Enum.any?(frames, fn + {:binary, bin} -> match?({:ok, {:auth_ok, _}, _}, TunnelFrame.decode(bin)) + _ -> false + end) + end + + test "one-stream golden path: agent replies headers+body", %{tunnel: tunnel} do + assert {:ok, ticket, _} = Connector.mint_session_ticket(tunnel) + assert {:push, _frames, state} = TunnelWebSock.init(%{session_ticket: ticket}) + + parent = self() + # Drive WebSock callbacks on this process by handling messages inline. + # Register a dedicated agent process that implements the open_stream protocol. + agent = + spawn_link(fn -> + receive do + {:open_stream, stream_id, _request, waiter, ref} -> + # Feed response frames through TunnelWebSock state machine + # We re-use a copy of state held in this process: + send(parent, {:opened, stream_id, waiter, ref}) + end + end) + + :ok = Connector.register_ws(tunnel.id, agent, %{}) + + origin = %{ + type: :tunnel, + tunnel_id: tunnel.id, + host: "app.example.com", + origin_host_header: "app.example.com" + } + + conn = + Phoenix.ConnTest.build_conn(:get, "/hello") + |> Map.put(:host, "app.example.com") + + task = + Task.async(fn -> + Connector.dispatch(conn, origin) + end) + + assert_receive {:opened, stream_id, waiter, ref}, 1_000 + + # Simulate agent → edge STREAM_HEADERS + STREAM_BODY by calling handle_in + # on a websock state that has the stream registered. Easier: send complete + # message directly to waiter as the connector protocol expects. + send( + waiter, + {:stream_complete, ref, stream_id, + %{status: 200, headers: [{"content-type", "text/plain"}], body: "ok from origin"}} + ) + + assert {:ok, %Plug.Conn{status: 200, resp_body: "ok from origin", halted: true}} = + Task.await(task, 2_000) + + # Keep state referenced so AUTH registration is meaningful + assert state.tunnel_id == tunnel.id + end + + test "dispatch returns error when agent is down", %{tunnel: tunnel} do + origin = %{type: :tunnel, tunnel_id: tunnel.id, host: "app.example.com"} + + conn = + Phoenix.ConnTest.build_conn(:get, "/hello") + |> Map.put(:host, "app.example.com") + + assert {:error, :agent_down} = Connector.dispatch(conn, origin) + end + + test "terminate fails waiters with stream_id without waiting for dispatch timeout", %{ + tunnel: tunnel + } do + assert {:ok, ticket, _} = Connector.mint_session_ticket(tunnel) + assert {:push, _frames, state} = TunnelWebSock.init(%{session_ticket: ticket}) + + parent = self() + stream_id = 99 + ref = make_ref() + + state = %{ + state + | streams: %{ + stream_id => %{ + stream_id: stream_id, + waiter: parent, + ref: ref, + status: nil, + headers: [], + body: <<>>, + body_bytes: 0 + } + } + } + + assert :ok = TunnelWebSock.terminate(:shutdown, state) + assert_receive {:stream_error, ^ref, ^stream_id, :agent_down}, 200 + end + + test "open_stream chunks large request bodies under frame max payload", %{tunnel: tunnel} do + assert {:ok, ticket, _} = Connector.mint_session_ticket(tunnel) + assert {:push, _frames, state} = TunnelWebSock.init(%{session_ticket: ticket}) + + # ~1.5 MiB body forces more than one STREAM_BODY frame (1 MiB payload cap). + large = :binary.copy(<<"x">>, TunnelFrame.max_payload_bytes() + 100_000) + waiter = self() + ref = make_ref() + + request = %{ + method: "POST", + path: "/upload", + headers: [{"content-type", "application/octet-stream"}], + body: large, + host: "app.example.com" + } + + assert {:push, frames, _state} = + TunnelWebSock.handle_info({:open_stream, 1, request, waiter, ref}, state) + + binaries = + Enum.map(frames, fn + {:binary, bin} -> bin + other -> flunk("unexpected frame: #{inspect(other)}") + end) + + joined = IO.iodata_to_binary(binaries) + {decoded, ""} = TunnelFrame.decode_all(joined) + + assert match?({:stream_open, _}, hd(decoded)) + body_frames = Enum.filter(decoded, &match?({:stream_body, _}, &1)) + assert length(body_frames) >= 2 + + reassembled = + body_frames + |> Enum.map(fn {:stream_body, %{data: d}} -> d end) + |> IO.iodata_to_binary() + + assert byte_size(reassembled) == byte_size(large) + assert List.last(body_frames) |> elem(1) |> Map.get(:fin) == true + end + + test "failed AUTH closes with stop reply", %{tunnel: _tunnel} do + assert {:stop, :normal, frames, state} = + TunnelWebSock.init(%{session_ticket: "ets_invalid_not_minted"}) + + assert state.closing + + assert Enum.any?(frames, fn + {:binary, bin} -> match?({:ok, {:auth_fail, _}, _}, TunnelFrame.decode(bin)) + _ -> false + end) + end + + test "DNSEdgeProxy maps request_too_large to 413 not 502", %{tunnel: tunnel} do + previous = Application.get_env(:elektrine, :dns, []) + + Application.put_env( + :elektrine, + :dns, + Keyword.merge(previous, + tunnel_enabled: true, + edge_proxy_enabled: true, + edge_proxy_origin_resolver: fn + "big.tunnel.test" -> + {:ok, + %{ + type: :tunnel, + tunnel_id: tunnel.id, + host: "big.tunnel.test", + origin_host_header: "big.tunnel.test", + atomine_gate: false + }} + + _ -> + {:error, :not_found} + end, + edge_proxy_tunnel_dispatcher: fn _conn, _origin -> + {:error, :request_too_large} + end + ) + ) + + on_exit(fn -> Application.put_env(:elektrine, :dns, previous) end) + + conn = + Phoenix.ConnTest.build_conn(:post, "http://big.tunnel.test/path") + |> Map.put(:host, "big.tunnel.test") + |> ElektrineWeb.Plugs.DNSEdgeProxy.call([]) + + assert conn.status == 413 + assert conn.halted + end + + test "DNSEdgeProxy returns 502 when tunnel agent is down", %{tunnel: tunnel} do + previous = Application.get_env(:elektrine, :dns, []) + + Application.put_env( + :elektrine, + :dns, + Keyword.merge(previous, + tunnel_enabled: true, + edge_proxy_enabled: true, + edge_proxy_origin_resolver: fn + "down.tunnel.test" -> + {:ok, + %{ + type: :tunnel, + tunnel_id: tunnel.id, + host: "down.tunnel.test", + origin_host_header: "down.tunnel.test", + atomine_gate: false + }} + + _ -> + {:error, :not_found} + end + ) + ) + + on_exit(fn -> Application.put_env(:elektrine, :dns, previous) end) + + conn = + Phoenix.ConnTest.build_conn(:get, "http://down.tunnel.test/path") + |> Map.put(:host, "down.tunnel.test") + |> ElektrineWeb.Plugs.DNSEdgeProxy.call([]) + + assert conn.status == 502 + assert conn.halted + end + + defp ensure_connector do + case Process.whereis(Connector) do + nil -> + start_supervised!(Connector) + + _pid -> + :ok + end + end +end diff --git a/apps/elektrine_web/test/elektrine_web/edge/tunnel_frame_test.exs b/apps/elektrine_web/test/elektrine_web/edge/tunnel_frame_test.exs new file mode 100644 index 0000000..ba0ef7f --- /dev/null +++ b/apps/elektrine_web/test/elektrine_web/edge/tunnel_frame_test.exs @@ -0,0 +1,102 @@ +defmodule ElektrineWeb.Edge.TunnelFrameTest do + use ExUnit.Case, async: true + + alias ElektrineWeb.Edge.TunnelFrame + + test "encodes and decodes AUTH / AUTH_OK round-trip" do + auth = %{session_ticket: "ets_abc", agent_version: "0.1.0", max_streams: 16} + assert {:ok, bin} = TunnelFrame.encode(:auth, auth) + assert {:ok, {:auth, decoded}, ""} = TunnelFrame.decode(bin) + assert decoded.session_ticket == "ets_abc" + assert decoded.max_streams == 16 + + ok = %{idle_timeout_ms: 60_000, server_time: "2026-01-01T00:00:00Z"} + assert {:ok, bin_ok} = TunnelFrame.encode(:auth_ok, ok) + assert {:ok, {:auth_ok, decoded_ok}, ""} = TunnelFrame.decode(bin_ok) + assert decoded_ok.idle_timeout_ms == 60_000 + end + + test "encodes and decodes STREAM_OPEN and STREAM_HEADERS" do + open = %{ + stream_id: 1, + method: "GET", + path: "/health?x=1", + headers: [["accept", "text/plain"]] + } + + assert {:ok, bin} = TunnelFrame.encode(:stream_open, open) + assert {:ok, {:stream_open, decoded}, ""} = TunnelFrame.decode(bin) + assert decoded.stream_id == 1 + assert decoded.method == "GET" + assert decoded.path == "/health?x=1" + + headers = %{stream_id: 1, status: 200, headers: [["content-type", "text/plain"]]} + assert {:ok, bin_h} = TunnelFrame.encode(:stream_headers, headers) + assert {:ok, {:stream_headers, decoded_h}, ""} = TunnelFrame.decode(bin_h) + assert decoded_h.status == 200 + end + + test "STREAM_BODY carries fin flag and raw bytes" do + assert {:ok, bin} = + TunnelFrame.encode(:stream_body, %{stream_id: 3, fin: true, data: "hello"}) + + assert {:ok, {:stream_body, body}, ""} = TunnelFrame.decode(bin) + assert body.stream_id == 3 + assert body.fin == true + assert body.data == "hello" + + assert {:ok, bin2} = + TunnelFrame.encode(:stream_body, %{stream_id: 3, fin: false, data: "part"}) + + assert {:ok, {:stream_body, body2}, ""} = TunnelFrame.decode(bin2) + assert body2.fin == false + end + + test "PING / PONG opaque 8 bytes" do + opaque = :crypto.strong_rand_bytes(8) + assert {:ok, bin} = TunnelFrame.encode(:ping, opaque) + assert {:ok, {:ping, ^opaque}, ""} = TunnelFrame.decode(bin) + + assert {:ok, bin2} = TunnelFrame.encode(:pong, opaque) + assert {:ok, {:pong, ^opaque}, ""} = TunnelFrame.decode(bin2) + end + + test "decode_all handles multiple frames and incomplete tail" do + f1 = TunnelFrame.encode!(:ping, <<1, 2, 3, 4, 5, 6, 7, 8>>) + f2 = TunnelFrame.encode!(:auth_fail, %{reason: "nope"}) + partial = binary_part(f2, 0, 3) + + assert {frames, rest} = TunnelFrame.decode_all(f1 <> partial) + assert length(frames) == 1 + assert {:ping, _} = hd(frames) + assert rest == partial + end + + test "rejects oversized payload length" do + huge = <> + assert {:error, :payload_too_large} = TunnelFrame.decode(huge) + end + + test "AUTH_FAIL and STREAM_RST JSON payloads" do + assert {:ok, bin} = TunnelFrame.encode(:auth_fail, %{reason: "invalid_ticket"}) + assert {:ok, {:auth_fail, %{reason: "invalid_ticket"}}, ""} = TunnelFrame.decode(bin) + + assert {:ok, bin2} = TunnelFrame.encode(:stream_rst, %{stream_id: 7, error: "timeout"}) + assert {:ok, {:stream_rst, decoded}, ""} = TunnelFrame.decode(bin2) + assert decoded.stream_id == 7 + assert decoded.error == "timeout" + end + + test "stream_body_max_data_bytes leaves room for stream_id and flags" do + assert TunnelFrame.stream_body_max_data_bytes() == TunnelFrame.max_payload_bytes() - 5 + + data = :binary.copy(<<"z">>, TunnelFrame.stream_body_max_data_bytes()) + assert {:ok, bin} = TunnelFrame.encode(:stream_body, %{stream_id: 1, fin: true, data: data}) + assert {:ok, {:stream_body, %{data: ^data, fin: true}}, ""} = TunnelFrame.decode(bin) + + too_big = data <> "x" + + assert {:error, :payload_too_large} = + TunnelFrame.encode(:stream_body, %{stream_id: 1, fin: true, data: too_big}) + end +end diff --git a/config/config.exs b/config/config.exs index 65e5a1c..db3a7ba 100644 --- a/config/config.exs +++ b/config/config.exs @@ -335,6 +335,14 @@ config :elektrine, :dns, edge_access_cookie_max_age_seconds: 8 * 60 * 60, edge_access_exchange_ttl_seconds: 60, edge_access_state_max_age_seconds: 600, +tunnel_enabled: false, + tunnel_max_per_zone: 10, + tunnel_max_streams: 16, + tunnel_idle_timeout_ms: 60_000, + tunnel_stream_idle_timeout_ms: 120_000, + tunnel_session_ttl_seconds: 3_600, + tunnel_connector_urls: [], + tunnel_dispatch_timeout_ms: 30_000, recursive_cache_max_entries: 10_000, recursive_cache_cleanup_interval_ms: 60_000, recursive_root_hints: [ diff --git a/config/runtime/dns.exs b/config/runtime/dns.exs index 06820e2..a904928 100644 --- a/config/runtime/dns.exs +++ b/config/runtime/dns.exs @@ -161,6 +161,19 @@ dns_public_bind_ip = nil -> Keyword.get(dns_config, :public_bind_ip) "" -> nil value -> String.trim(value) +dns_tunnel_connector_urls = + case System.get_env("DNS_TUNNEL_CONNECTOR_URLS") do + nil -> + Keyword.get(dns_config, :tunnel_connector_urls, []) + + "" -> + [] + + value -> + value + |> String.split(",", trim: true) + |> Enum.map(&String.trim/1) + |> Enum.reject(&(&1 == "")) end config :elektrine, :dns, @@ -243,6 +256,22 @@ config :elektrine, :dns, "DNS_QUERY_STATS_RETENTION_MAX_BATCHES", Keyword.get(dns_config, :query_stats_retention_max_batches, 100) ), +tunnel_enabled: + parse_bool_env.( + "DNS_TUNNEL_ENABLED", + Keyword.get(dns_config, :tunnel_enabled, false) + ), + tunnel_max_streams: + parse_int_env.( + "DNS_TUNNEL_MAX_STREAMS", + Keyword.get(dns_config, :tunnel_max_streams, 16) + ), + tunnel_max_per_zone: + parse_int_env.( + "DNS_TUNNEL_MAX_PER_ZONE", + Keyword.get(dns_config, :tunnel_max_per_zone, 10) + ), + tunnel_connector_urls: dns_tunnel_connector_urls, max_udp_payload: parse_int_env.( "DNS_MAX_UDP_PAYLOAD", diff --git a/docs/api/external-api.md b/docs/api/external-api.md index 734b849..04bcaf0 100644 --- a/docs/api/external-api.md +++ b/docs/api/external-api.md @@ -30,6 +30,7 @@ rejected with `403`. Valid scopes: | Nerve | `read:nerve` | `write:nerve` | | Kairo | `read:kairo` | `write:kairo` | | DNS | `read:dns` | `write:dns` | +| Edge | `read:edge` | `write:edge` | | Proofs | `read:proofs` | `write:proofs` | | Static site | `read:static_site` | `write:static_site` | | Moderation | `read:moderation` | `write:moderation` | @@ -187,6 +188,22 @@ analytics responses accordingly. other external API controllers. - Response echoes `window_days` / `window_hours` and `retention_days`. +### Edge tunnels (`read:edge` / `write:edge`) + +Private-origin tunnels for the edge proxy. Minting returns the long-lived agent +token (`etn_...`) **once**. A `write:dns`-only PAT cannot mint or revoke tunnels. + +| Endpoint | Scope | Description | +| --- | --- | --- | +| `GET /dns/zones/:zone_id/tunnels` | `read:edge` | List tunnels for a zone (no secrets) | +| `GET /dns/zones/:zone_id/tunnels/:id` | `read:edge` | Show one tunnel | +| `POST /dns/zones/:zone_id/tunnels` | `write:edge` | Mint a tunnel (`name`, optional `allowed_hosts`) | +| `PUT /dns/zones/:zone_id/tunnels/:id` | `write:edge` | Update name / allowed hosts | +| `DELETE /dns/zones/:zone_id/tunnels/:id` | `write:edge` | Revoke a tunnel | + +Agent control-plane paths under `/_edge/tunnel/v1/*` use the tunnel token, not a +user PAT. Feature flag: `DNS_TUNNEL_ENABLED` (default `false`). + ### Kairo (`read:kairo` / `write:kairo`) | Endpoint | Description | diff --git a/docs/architecture/dns-module.md b/docs/architecture/dns-module.md index c08aa6f..334e91e 100644 --- a/docs/architecture/dns-module.md +++ b/docs/architecture/dns-module.md @@ -113,6 +113,10 @@ PoPs. Rate-limit defaults live in `config/runtime/dns.exs`. - planned: `read:edge` / `write:edge` for tunnel mint/revoke and cache purge (see [edge-platform.md](edge-platform.md)); zone/record/service scopes stay on `read:dns` / `write:dns` +- `read:dns` lists zones and records +- `write:dns` creates, updates, verifies, and deletes zones and records +- `read:edge` lists private-origin tunnels for a zone +- `write:edge` mints, updates, and revokes tunnels (token shown once); reserved for edge cache purge Current PAT endpoints: @@ -149,6 +153,11 @@ Query analytics: `Elektrine.Profiles.AnalyticsRetentionWorker` - Stored `qname` values can be sensitive; operators should size retention accordingly +- `GET /api/ext/v1/dns/zones/:zone_id/tunnels` (`read:edge`) +- `GET /api/ext/v1/dns/zones/:zone_id/tunnels/:id` (`read:edge`) +- `POST /api/ext/v1/dns/zones/:zone_id/tunnels` (`write:edge`) +- `PUT /api/ext/v1/dns/zones/:zone_id/tunnels/:id` (`write:edge`) +- `DELETE /api/ext/v1/dns/zones/:zone_id/tunnels/:id` (`write:edge`) Internal endpoints: @@ -158,6 +167,39 @@ Internal endpoints: - TLS allow-list and origin lookup under `/_edge/*` (Caddy on-demand TLS and external data-plane bridge) — see [caddy.md](../self-hosting/caddy.md) and edge-platform +- `POST /_edge/tunnel/v1/register` exchanges tunnel token (`etn_...`) for a session ticket +- `POST /_edge/tunnel/v1/refresh` rotates the session ticket +- `POST /_edge/tunnel/v1/disconnect` tears down the agent session +- `GET /_edge/tunnel/v1/ws` WSS upgrade to `ElektrineWeb.Edge.Connector` (session ticket) + +### Private origin tunnels (feature-flagged) + +Default **off** via `DNS_TUNNEL_ENABLED=false`. When enabled, the edge node **must** +run with web enabled (`ELEKTRINE_ENABLE_WEB=true`) so Phoenix serves +`/_edge/tunnel/v1/*` and `DNSEdgeProxy` can dispatch tunnel origins. + +- Table `dns_tunnels` stores zone-owned credentials (token hashed, shown once as `etn_...`). +- Proxied records with `origin_type=tunnel` use content `tunnel.invalid` (CNAME/ALIAS only) + and a zone-owned `tunnel_id` (see PR 7a validation). +- Agents register over HTTPS, then open outbound WSS; the edge never dials private URLs. + The agent dials only operator-configured `ORIGIN_URL`. +- Wire protocol: 4-byte big-endian length + 1-byte type + payload (`ElektrineWeb.Edge.TunnelFrame`). +- Packaging: `mix elektrine.tunnel_agent` or `scripts/edge/elektrine-tunnel-agent.sh` + (WebSockex client, pinned in `mix.lock`). +- Session tickets and WSS sessions are **local to one BEAM node** (ETS + GenServer). + Register and WSS must hit the **same** edge instance (sticky routing or a single + connector URL). Multi-node ticket fan-out is out of scope for v1. +- Agent authenticates with an AUTH frame (session ticket is not put on the WSS + query string by default; query still accepted server-side for compatibility). +- Agent HTTP to `ORIGIN_URL` uses `autoredirect: false` so local origins cannot + bounce the agent at off-host URLs. +- `stream_idle_timeout_ms` is returned at register for forward compatibility; v1 + enforces wait via `tunnel_dispatch_timeout_ms` on the edge waiter only. +- Revoke drops live WSS + tickets on the **local** connector node (best-effort). + +Related env keys: `DNS_TUNNEL_ENABLED`, `DNS_TUNNEL_MAX_STREAMS` (default 16), +`DNS_TUNNEL_MAX_PER_ZONE` (default 10), `DNS_TUNNEL_CONNECTOR_URLS` (optional WSS URLs; +prefer a single sticky connector host). ## Docker deployment notes diff --git a/env/presets/dns.env b/env/presets/dns.env index 28124f6..49ce437 100644 --- a/env/presets/dns.env +++ b/env/presets/dns.env @@ -13,6 +13,12 @@ # set only on /_edge/access/v1/complete for the customer host; no parent-domain cookie. # DNS_EDGE_ACCESS_EXCHANGE_TTL_SECONDS=60 # DNS_EDGE_ACCESS_COOKIE_MAX_AGE_SECONDS=28800 +# Private origin tunnels (edge agent). Default off. Requires web-enabled node +# (not stock DNS worker). See scripts/edge/ and docs/architecture/dns-module.md. +# DNS_TUNNEL_ENABLED=false +# DNS_TUNNEL_MAX_STREAMS=16 +# DNS_TUNNEL_MAX_PER_ZONE=10 +# DNS_TUNNEL_CONNECTOR_URLS=wss://edge.example.com/_edge/tunnel/v1/ws # If this host also runs NetBird or another private DNS listener, bind public # authoritative DNS to the public interface instead of 0.0.0.0. diff --git a/mix.lock b/mix.lock index fd7bae9..be89cde 100644 --- a/mix.lock +++ b/mix.lock @@ -91,6 +91,7 @@ "web_driver_client": {:hex, :web_driver_client, "0.3.0", "25c53ffdeb779ff933c7cc145b71229662421078c10dbc43b3afab846e65366c", [:mix], [{:hackney, "~> 1.6 or ~> 4.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:tesla, "~> 1.3", [hex: :tesla, repo: "hexpm", optional: false]}], "hexpm", "f89a43505b4fa5e1d5dc50818980a4d2703e390dca29fa0ad39911b0eb46b65a"}, "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"}, "websock_adapter": {:hex, :websock_adapter, "0.6.0", "73db5ab8aaefd1a876a97ce3e6afc96562625de69ef17a4e04426e034849d0b8", [:mix], [{:bandit, ">= 0.6.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "50021a85bce8f203b086705d9e0c5415e2c7eb05d319111b0428fe71f9934617"}, + "websockex": {:hex, :websockex, "0.5.1", "9de28d37bbe34f371eb46e29b79c94c94fff79f93c960d842fbf447253558eb4", [:mix], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8ef39576ed56bc3804c9cd8626f8b5d6b5721848d2726c0ccd4f05385a3c9f14"}, "webtransport": {:hex, :webtransport, "0.4.3", "df0c53da138cdc5f3390a31ef612caa3a041c23e18c1a0b6fbd4d72743277806", [:rebar3], [{:h2, "~> 0.10.4", [hex: :h2, repo: "hexpm", optional: false]}, {:quic, "~> 1.7.0", [hex: :quic, repo: "hexpm", optional: false]}], "hexpm", "3ae8c76696cdb56bee3caf1dd853a0a596339886468ca17c578a9bfd647dab9d"}, "x509": {:hex, :x509, "0.9.2", "a75aa605348abd905990f3d2dc1b155fcde4e030fa2f90c4a91534405dce0f6e", [:mix], [], "hexpm", "4c5ede75697e565d4b0f5be04c3b71bb1fd3a090ea243af4bd7dae144e48cfc7"}, "yamerl": {:hex, :yamerl, "0.10.0", "4ff81fee2f1f6a46f1700c0d880b24d193ddb74bd14ef42cb0bcf46e81ef2f8e", [:rebar3], [], "hexpm", "346adb2963f1051dc837a2364e4acf6eb7d80097c0f53cbdc3046ec8ec4b4e6e"}, diff --git a/scripts/check_maintainability_budgets.sh b/scripts/check_maintainability_budgets.sh index 241af3e..708a827 100755 --- a/scripts/check_maintainability_budgets.sh +++ b/scripts/check_maintainability_budgets.sh @@ -45,7 +45,7 @@ check_max_lines_matching() { # the largest files from growing while gradual extraction continues. check_max_lines config/runtime.exs 1210 check_max_lines config/runtime/bluesky.exs 75 -check_max_lines config/runtime/dns.exs 250 +check_max_lines config/runtime/dns.exs 280 check_max_lines config/runtime/messaging_federation.exs 125 check_max_lines config/runtime/uploads.exs 100 check_max_lines config/runtime/webrtc.exs 125 diff --git a/scripts/edge/README.md b/scripts/edge/README.md new file mode 100644 index 0000000..89f8a27 --- /dev/null +++ b/scripts/edge/README.md @@ -0,0 +1,38 @@ +# Edge tunnel agent + +Outbound agent that publishes a private HTTP origin through Elektrine's edge +proxy. The edge never dials the origin; the agent dials only `ORIGIN_URL`. + +## Requirements + +- Elektrine edge node with `ELEKTRINE_ENABLE_WEB=true` and `DNS_TUNNEL_ENABLED=true` +- Zone tunnel minted with a `write:edge` PAT +- Local HTTP service at `ORIGIN_URL` + +## Quick start + +1. Mint a tunnel (token shown once): + + ```bash + curl -sS -X POST \ + -H "Authorization: Bearer ekt_..." \ + -H "Content-Type: application/json" \ + -d '{"name":"home-lab","allowed_hosts":["app.example.com"]}' \ + "https://example.com/api/ext/v1/dns/zones/ZONE_ID/tunnels" + ``` + +2. Bind a proxied CNAME/ALIAS with `origin_type=tunnel`, content `tunnel.invalid`, + and the tunnel id. + +3. Run the agent: + + ```bash + cp scripts/edge/elektrine-tunnel-agent.env.example /etc/elektrine/tunnel-agent.env + # edit env + scripts/edge/elektrine-tunnel-agent.sh + ``` + +## Protocol + +See `docs/architecture/dns-module.md` (tunnel section). Frames are length-prefixed +binary over WSS; v1 does not support WebSocket/SSE/CONNECT through the tunnel. diff --git a/scripts/edge/elektrine-tunnel-agent.env.example b/scripts/edge/elektrine-tunnel-agent.env.example new file mode 100644 index 0000000..a8d5fab --- /dev/null +++ b/scripts/edge/elektrine-tunnel-agent.env.example @@ -0,0 +1,13 @@ +# Control plane base URL (HTTPS). No trailing slash. +CONTROL_PLANE_URL=https://edge.example.com + +# Long-lived tunnel token from POST /api/ext/v1/dns/zones/:id/tunnels (write:edge). +# Shown once at mint; prefix etn_ +TUNNEL_TOKEN=etn_replace_me + +# Local origin the agent may dial. Operator-controlled only — never taken from edge frames. +ORIGIN_URL=http://127.0.0.1:8080 + +# Optional +# AGENT_VERSION=elektrine-tunnel-agent/0.1.0 +# MAX_STREAMS=16 diff --git a/scripts/edge/elektrine-tunnel-agent.sh b/scripts/edge/elektrine-tunnel-agent.sh new file mode 100755 index 0000000..6d1106f --- /dev/null +++ b/scripts/edge/elektrine-tunnel-agent.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +# Bootstrap wrapper for the Elektrine edge tunnel agent MVP. +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "$ROOT" + +if [[ -f "${1:-}" ]]; then + set -a + # shellcheck disable=SC1090 + source "$1" + set +a + shift +elif [[ -f /etc/elektrine/tunnel-agent.env ]]; then + set -a + # shellcheck disable=SC1091 + source /etc/elektrine/tunnel-agent.env + set +a +fi + +: "${CONTROL_PLANE_URL:?CONTROL_PLANE_URL is required}" +: "${TUNNEL_TOKEN:?TUNNEL_TOKEN is required}" +: "${ORIGIN_URL:?ORIGIN_URL is required}" + +export CONTROL_PLANE_URL TUNNEL_TOKEN ORIGIN_URL +export AGENT_VERSION="${AGENT_VERSION:-elektrine-tunnel-agent/0.1.0}" + +exec mix elektrine.tunnel_agent "$@"