elektrine/elektrine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

merge: pr-7b-featdns-tunnel-control-plane-wss-connector-agent-m

Browse source
This commit is contained in:
maxfield 2026-08-02 01:54:36 -04:00
commit 008ecba84d
39 changed files with 3728 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

15
apps/elektrine/lib/elektrine/application.ex
View file

@ -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

11
apps/elektrine/lib/elektrine/developer/api_token.ex
View file

@ -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"}

1
apps/elektrine/mix.exs
View file

@ -67,6 +67,7 @@ defmodule Elektrine.MixProject do
Elektrine.Social.PostBoost,
Elektrine.Social.PostLike,
ElektrineWeb.Endpoint,
ElektrineWeb.Edge.Connector,
ElektrineEmailWeb.HarakaWebhookController,
ElektrineWeb.Presence
]

25
apps/elektrine/priv/repo/migrations/20260802010000_create_dns_tunnels.exs Normal file
View file

@ -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

75
apps/elektrine/priv/static/openapi/ext-v1.yaml
View file

@ -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

20
apps/elektrine_dns/lib/elektrine/dns.ex
View file

@ -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(".")

185
apps/elektrine_dns/lib/elektrine/dns/tunnel.ex Normal file
View file

@ -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

488
apps/elektrine_dns/lib/elektrine/dns/tunnel_agent.ex Normal file
View file

@ -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

70
apps/elektrine_dns/lib/elektrine/dns/tunnel_agent/ws_client.ex Normal file
View file

@ -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

249
apps/elektrine_dns/lib/elektrine/dns/tunnels.ex Normal file
View file

@ -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

1
apps/elektrine_dns/lib/elektrine/dns/zone.ex
View file

@ -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

127
apps/elektrine_dns/lib/elektrine_dns_web/controllers/api/dns_controller.ex
View file

@ -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

25
apps/elektrine_dns/lib/mix/tasks/elektrine.tunnel_agent.ex Normal file
View file

@ -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

4
apps/elektrine_dns/mix.exs
View file

@ -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

143
apps/elektrine_dns/test/elektrine/dns/tunnel_agent_test.exs Normal file
View file

@ -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

119
apps/elektrine_dns/test/elektrine/dns/tunnels_test.exs Normal file
View file

@ -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

90
apps/elektrine_social/lib/elektrine_social_web/live/timeline_live/operations/post_operations.ex
View file

@ -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 ->

202
apps/elektrine_web/lib/elektrine_web/controllers/tunnel_controller.ex Normal file
View file

@ -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

394
apps/elektrine_web/lib/elektrine_web/edge/connector.ex Normal file
View file

@ -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

209
apps/elektrine_web/lib/elektrine_web/edge/tunnel_frame.ex Normal file
View file

@ -0,0 +1,209 @@
defmodule ElektrineWeb.Edge.TunnelFrame do
@moduledoc """
Length-prefixed binary frame codec for the edge tunnel WSS protocol (v1).
Wire format:
<<payload_len::32-big, type::8, payload::binary-size(payload_len)>>
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 <<stream_id::32, flags::8, data::binary>>
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, <<byte_size(body)::32-big, type::8, body::binary>>}
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(<<payload_len::32-big, _rest::binary>>)
when payload_len > @max_payload_bytes do
{:error, :payload_too_large}
end
def decode(<<payload_len::32-big, type::8, rest::binary>> = 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
<<stream_id::32-big, flags::8, data::binary>>
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, <<stream_id::32-big, flags::8, data::binary>>) 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

442
apps/elektrine_web/lib/elektrine_web/edge/tunnel_websock.ex Normal file
View file

@ -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

26
apps/elektrine_web/lib/elektrine_web/plugs/dns_edge_proxy.ex
View file

@ -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
[

29
apps/elektrine_web/lib/elektrine_web/router.ex
View file

@ -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])

19
apps/elektrine_web/lib/elektrine_web/routes/dns.ex
View file

@ -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

2
apps/elektrine_web/mix.exs
View file

@ -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,

86
apps/elektrine_web/test/elektrine_web/controllers/api/dns_tunnels_controller_test.exs Normal file
View file

@ -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

96
apps/elektrine_web/test/elektrine_web/controllers/tunnel_controller_test.exs Normal file
View file

@ -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

305
apps/elektrine_web/test/elektrine_web/edge/connector_test.exs Normal file
View file

@ -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

102
apps/elektrine_web/test/elektrine_web/edge/tunnel_frame_test.exs Normal file
View file

@ -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 = <<TunnelFrame.max_payload_bytes() + 1::32-big, 0x01, 0>>
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

8
config/config.exs
View file

@ -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: [

29
config/runtime/dns.exs
View file

@ -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",

17
docs/api/external-api.md
View file

@ -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 |

42
docs/architecture/dns-module.md
View file

@ -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

6
env/presets/dns.env vendored
View file

@ -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.

1
mix.lock
View file

@ -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"},

2
scripts/check_maintainability_budgets.sh
View file

@ -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

38
scripts/edge/README.md Normal file
View file

@ -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.

13
scripts/edge/elektrine-tunnel-agent.env.example Normal file
View file

@ -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

28
scripts/edge/elektrine-tunnel-agent.sh Executable file
View file

@ -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 "$@"