lfg2025/archy
1
0
Fork 0
Code Issues 27 Pull Requests 5 Actions Packages Projects Releases 6 Wiki Activity

Archipelago — open-source initial import

Browse Source
This commit is contained in:
Archipelago
2026-08-12 10:55:49 +00:00
commit c7930b6e38
1578 changed files with 332819 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/handler/blob.rs
+234
View File
@@ -0,0 +1,234 @@
//! HTTP handlers for the content-addressed blob store.
//!
//! - `POST /api/blob` — session-authenticated. Raw body is the blob;
//! headers set mime/filename. Returns `{cid, size, mime}`.
//! - `GET /blob/<cid>?cap=<hex>&exp=<epoch>&peer=<pubkey>` — peer-facing.
//! Capability verified against the stored HMAC key; bytes streamed back.
use super::{build_response, ApiHandler};
use crate::blobs::BlobStore;
use anyhow::Result;
use hyper::{Body, HeaderMap, Response, StatusCode};
use std::path::Path;
use std::sync::Arc;
/// Read the archipelago .onion address if Tor has published one, so uploads
/// that need to be publicly reachable (profile pictures, banners) can return
/// a URL a peer outside the LAN can actually fetch. Returns `None` before
/// onboarding or when Tor isn't running — callers fall back to the local
/// self-test URL.
async fn read_self_onion(data_dir: &Path) -> Option<String> {
let hostnames = data_dir.join("tor-hostnames").join("archipelago");
let legacy = Path::new("/var/lib/archipelago/tor-hostnames/archipelago");
for p in [hostnames.as_path(), legacy] {
if let Ok(s) = tokio::fs::read_to_string(p).await {
let trimmed = s.trim();
if !trimmed.is_empty() {
return Some(trimmed.to_string());
}
}
}
None
}
impl ApiHandler {
pub(super) async fn handle_blob_upload(
store: &Arc<BlobStore>,
self_pubkey_hex: &str,
data_dir: &Path,
headers: &HeaderMap,
body: hyper::body::Bytes,
) -> Result<Response<Body>> {
let mime = headers
.get("x-blob-mime")
.and_then(|v| v.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let filename = headers
.get("x-blob-filename")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
// Optional caller-supplied thumbnail (small, base64) — e.g. the mesh
// chat's image-quality picker generates a tiny client-side preview so
// a ContentRef receiver can render something before fetching the full
// blob. Best-effort: a malformed header is just ignored, not fatal.
let thumb_bytes = headers
.get("x-blob-thumb")
.and_then(|v| v.to_str().ok())
.and_then(|b64| {
use base64::{engine::general_purpose::STANDARD, Engine as _};
STANDARD.decode(b64).ok()
});
let bytes = body.to_vec();
// Uploads through /api/blob come from the node owner's session and
// are almost always intended for external consumption (profile
// pictures, banners). Store them public so `/blob/<cid>` serves
// without a capability check — external Nostr clients fetching a
// kind-0 `picture` URL have no cap and can't get one.
match store.put(&bytes, &mime, filename, thumb_bytes, true).await {
Ok(meta) => {
let exp =
(chrono::Utc::now().timestamp() as u64) + crate::blobs::DEFAULT_CAP_TTL_SECS;
let cap = store.issue_capability(&meta.cid, self_pubkey_hex, exp);
let self_test_url = format!(
"/blob/{}?cap={}&exp={}&peer={}",
meta.cid, cap, exp, self_pubkey_hex
);
let public_url = match read_self_onion(data_dir).await {
Some(onion) => format!("http://{}/blob/{}", onion, meta.cid),
// Pre-onboarding / Tor-not-up: surface the local path so
// the UI doesn't break; publishing to Nostr should wait
// until Tor is live anyway.
None => format!("/blob/{}", meta.cid),
};
let resp = serde_json::json!({
"cid": meta.cid,
"size": meta.size,
"mime": meta.mime,
"filename": meta.filename,
"public_url": public_url,
"self_test_url": self_test_url,
});
Ok(build_response(
StatusCode::OK,
"application/json",
Body::from(serde_json::to_vec(&resp).unwrap_or_default()),
))
}
Err(e) => Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
Body::from(format!("blob upload failed: {}", e)),
)),
}
}
/// Share-to-mesh iframe intent. Mirrors `handle_blob_upload` but adds
/// CORS headers for the requesting app origin and returns a small JSON
/// payload the app forwards to its parent via postMessage:
/// `{ type: "share-to-mesh", cid, size, mime, filename }`.
pub(super) async fn handle_share_to_mesh(
store: &Arc<BlobStore>,
self_pubkey_hex: &str,
headers: &HeaderMap,
body: hyper::body::Bytes,
origin: &str,
) -> Result<Response<Body>> {
let mime = headers
.get("x-blob-mime")
.and_then(|v| v.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let filename = headers
.get("x-blob-filename")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let bytes = body.to_vec();
let meta = match store.put(&bytes, &mime, filename, None, false).await {
Ok(m) => m,
Err(e) => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
Body::from(format!("share-to-mesh failed: {}", e)),
));
}
};
// Self-signed capability so the app can preview/download its own
// upload before the user has picked a peer.
let exp = (chrono::Utc::now().timestamp() as u64) + crate::blobs::DEFAULT_CAP_TTL_SECS;
let cap = store.issue_capability(&meta.cid, self_pubkey_hex, exp);
let self_url = format!(
"/blob/{}?cap={}&exp={}&peer={}",
meta.cid, cap, exp, self_pubkey_hex
);
let resp = serde_json::json!({
"type": "share-to-mesh",
"cid": meta.cid,
"size": meta.size,
"mime": meta.mime,
"filename": meta.filename,
"self_url": self_url,
});
let body_vec = serde_json::to_vec(&resp).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.header("Access-Control-Allow-Origin", origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
.body(Body::from(body_vec))
.unwrap_or_else(|_| Response::new(Body::from("internal error"))))
}
pub(super) async fn handle_blob_download(
store: &Arc<BlobStore>,
path: &str,
query: &str,
) -> Result<Response<Body>> {
let cid = path.strip_prefix("/blob/").unwrap_or("");
if cid.is_empty() || !cid.chars().all(|c| c.is_ascii_hexdigit()) || cid.len() != 64 {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
Body::from("invalid cid"),
));
}
// Public blobs (profile pictures, banners) bypass the capability
// check — their CID is published on Nostr relays where any reader
// can see it, and external readers have no way to obtain a cap.
// Only blobs explicitly marked public at upload time qualify.
let is_public = store.meta(cid).await.map(|m| m.public).unwrap_or(false);
if !is_public {
let mut cap = None;
let mut exp: Option<u64> = None;
let mut peer = None;
for pair in query.split('&') {
let mut it = pair.splitn(2, '=');
match (it.next(), it.next()) {
(Some("cap"), Some(v)) => cap = Some(v.to_string()),
(Some("exp"), Some(v)) => exp = v.parse().ok(),
(Some("peer"), Some(v)) => peer = Some(v.to_string()),
_ => {}
}
}
let (Some(cap), Some(exp), Some(peer)) = (cap, exp, peer) else {
return Ok(build_response(
StatusCode::UNAUTHORIZED,
"text/plain",
Body::from("missing cap/exp/peer"),
));
};
if let Err(e) = store.verify_capability(cid, &peer, exp, &cap) {
tracing::warn!("blob cap rejected: cid={} peer={} reason={}", cid, peer, e);
return Ok(build_response(
StatusCode::FORBIDDEN,
"text/plain",
Body::from(format!("capability rejected: {}", e)),
));
}
}
let bytes = match store.get(cid).await {
Ok(b) => b,
Err(_) => {
return Ok(build_response(
StatusCode::NOT_FOUND,
"text/plain",
Body::from("blob not found"),
))
}
};
let mime = store
.meta(cid)
.await
.map(|m| m.mime)
.unwrap_or_else(|_| "application/octet-stream".to_string());
Ok(build_response(StatusCode::OK, &mime, Body::from(bytes)))
}
}
core/archipelago/src/api/handler/content.rs
+479
View File
@@ -0,0 +1,479 @@
use super::build_response;
use crate::config::Config;
use crate::content_server;
use anyhow::Result;
use hyper::{Response, StatusCode};
use super::{is_valid_app_id, ApiHandler};
impl ApiHandler {
pub(super) async fn handle_content_catalog(config: &Config) -> Result<Response<hyper::Body>> {
match content_server::load_catalog(&config.data_dir).await {
Ok(catalog) => {
// Only expose public metadata for available items
let items: Vec<serde_json::Value> = catalog
.items
.iter()
.filter(|i| !matches!(i.availability, content_server::Availability::Nobody))
.map(|i| {
serde_json::json!({
"id": i.id,
"filename": i.filename,
"mime_type": i.mime_type,
"size_bytes": i.size_bytes,
"description": i.description,
"access": i.access,
})
})
.collect();
let body =
serde_json::to_vec(&serde_json::json!({ "items": items })).unwrap_or_default();
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(body),
))
}
Err(e) => {
let body = serde_json::json!({ "error": e.to_string() });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Ok(build_response(
StatusCode::INTERNAL_SERVER_ERROR,
"application/json",
hyper::Body::from(body_bytes),
))
}
}
}
pub(super) async fn handle_content_request(
path: &str,
headers: &hyper::HeaderMap,
config: &Config,
) -> Result<Response<hyper::Body>> {
let content_id = path.strip_prefix("/content/").unwrap_or("");
if content_id.is_empty() || !is_valid_app_id(content_id) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid content ID"),
));
}
// Extract payment token from X-Payment-Token header
let payment_token = headers
.get("x-payment-token")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
// Extract a paid-entitlement gate token from X-Invoice-Hash (Lightning)
// or X-Onchain-Address (on-chain) — both authorize the download if this
// node issued+settled them, and both resolve against the same shared
// entitlement store keyed by the token string (#46).
let invoice_hash = headers
.get("x-invoice-hash")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string())
.or_else(|| {
headers
.get("x-onchain-address")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string())
});
// Extract federation peer DID from X-Federation-DID header
let peer_did = headers
.get("x-federation-did")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
// Parse Range header for streaming support
let range = headers
.get("range")
.and_then(|v| v.to_str().ok())
.and_then(content_server::parse_range_header);
match content_server::serve_content(
&config.data_dir,
content_id,
payment_token.as_deref(),
invoice_hash.as_deref(),
peer_did.as_deref(),
range,
)
.await
{
Ok(content_server::ServeResult::Ok(bytes, mime_type)) => {
let len = bytes.len();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", mime_type)
.header("Content-Length", len.to_string())
.header("Accept-Ranges", "bytes")
.body(hyper::Body::from(bytes))
.unwrap())
}
Ok(content_server::ServeResult::Partial {
bytes,
mime_type,
start,
end,
total,
}) => Ok(Response::builder()
.status(StatusCode::PARTIAL_CONTENT)
.header("Content-Type", mime_type)
.header("Content-Length", bytes.len().to_string())
.header(
"Content-Range",
format!("bytes {}-{}/{}", start, end, total),
)
.header("Accept-Ranges", "bytes")
.body(hyper::Body::from(bytes))
.unwrap()),
Ok(content_server::ServeResult::PaymentRequired(price_sats)) => {
let body = serde_json::json!({
"error": "Payment required",
"price_sats": price_sats,
"payment_header": "X-Payment-Token",
});
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Ok(build_response(
StatusCode::PAYMENT_REQUIRED,
"application/json",
hyper::Body::from(body_bytes),
))
}
Ok(content_server::ServeResult::Forbidden) => Ok(build_response(
StatusCode::FORBIDDEN,
"application/json",
hyper::Body::from(
r#"{"error":"This file is shared with the host's federation peers only. Federate with that node (exchange invites) so it recognizes you, then try again."}"#,
),
)),
Ok(content_server::ServeResult::NotFound) | Err(_) => Ok(build_response(
StatusCode::NOT_FOUND,
"text/plain",
hyper::Body::from("Content not found"),
)),
}
}
/// Seller side (#46): mint a Lightning invoice for a paid catalog item so a
/// buyer can pay from any external wallet. Path: GET /content/{id}/invoice.
/// Records a pending entitlement keyed by the invoice's payment hash.
pub(super) async fn handle_content_invoice(&self, path: &str) -> Result<Response<hyper::Body>> {
let content_id = path
.strip_prefix("/content/")
.and_then(|s| s.strip_suffix("/invoice"))
.unwrap_or("");
if content_id.is_empty() || !is_valid_app_id(content_id) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid content ID"),
));
}
let catalog = content_server::load_catalog(&self.config.data_dir)
.await
.unwrap_or_default();
let item = match catalog.items.iter().find(|i| i.id == content_id) {
Some(i) => i,
None => {
return Ok(build_response(
StatusCode::NOT_FOUND,
"text/plain",
hyper::Body::from("Content not found"),
))
}
};
let price_sats = match &item.access {
content_server::AccessControl::Paid { price_sats } => *price_sats,
_ => {
// Not a paid item — no invoice to issue.
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"Item is not paid"}"#),
));
}
};
let memo = format!("Archipelago peer file {content_id}");
match self
.rpc_handler
.create_invoice(price_sats as i64, &memo)
.await
{
Ok((bolt11, payment_hash)) if !payment_hash.is_empty() => {
crate::content_invoice::record_pending(&payment_hash, content_id, price_sats).await;
let body = serde_json::json!({
"bolt11": bolt11,
"payment_hash": payment_hash,
"price_sats": price_sats,
});
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
Ok(_) => Ok(build_response(
StatusCode::INTERNAL_SERVER_ERROR,
"application/json",
hyper::Body::from(r#"{"error":"Invoice missing payment hash"}"#),
)),
Err(e) => {
// Surface the FULL error chain ({:#}) — the generic top-level
// message hid the real cause (e.g. the LND REST connection
// failing), which made this 503 undiagnosable.
tracing::warn!("content invoice creation failed: {e:#}");
let body = serde_json::json!({
"error": format!("Could not create invoice: {e:#}")
});
Ok(build_response(
StatusCode::SERVICE_UNAVAILABLE,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
}
}
/// Seller side (#46): report whether a previously-issued invoice has settled.
/// Path: GET /content/{id}/invoice-status/{payment_hash}. On settlement the
/// entitlement is marked paid so the buyer can then download the file.
pub(super) async fn handle_content_invoice_status(
&self,
path: &str,
) -> Result<Response<hyper::Body>> {
let rest = path.strip_prefix("/content/").unwrap_or("");
let (content_id, payment_hash) = match rest.split_once("/invoice-status/") {
Some((id, hash)) => (id, hash),
None => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid request"),
))
}
};
if content_id.is_empty() || !is_valid_app_id(content_id) || payment_hash.is_empty() {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid request"),
));
}
// The hash must be one we issued for exactly this content item.
match crate::content_invoice::lookup(payment_hash).await {
Some((cid, _)) if cid == content_id => {}
_ => {
return Ok(build_response(
StatusCode::NOT_FOUND,
"application/json",
hyper::Body::from(r#"{"error":"Unknown invoice"}"#),
))
}
}
// Already paid? Otherwise ask our LND and persist the result.
let mut paid = crate::content_invoice::is_paid_for(payment_hash, content_id).await;
if !paid {
if let Ok(true) = self.rpc_handler.invoice_is_settled(payment_hash).await {
crate::content_invoice::mark_paid(payment_hash).await;
paid = true;
}
}
let body = serde_json::json!({ "paid": paid });
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
/// Seller side (#46): issue a fresh on-chain address for a paid catalog item
/// so a buyer can pay on-chain. Path: GET /content/{id}/onchain. Records a
/// pending entitlement keyed by the address; price doubles as expected amount.
pub(super) async fn handle_content_onchain(&self, path: &str) -> Result<Response<hyper::Body>> {
let content_id = path
.strip_prefix("/content/")
.and_then(|s| s.strip_suffix("/onchain"))
.unwrap_or("");
if content_id.is_empty() || !is_valid_app_id(content_id) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid content ID"),
));
}
let catalog = content_server::load_catalog(&self.config.data_dir)
.await
.unwrap_or_default();
let price_sats = match catalog.items.iter().find(|i| i.id == content_id) {
Some(i) => match &i.access {
content_server::AccessControl::Paid { price_sats } => *price_sats,
_ => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"Item is not paid"}"#),
))
}
},
None => {
return Ok(build_response(
StatusCode::NOT_FOUND,
"text/plain",
hyper::Body::from("Content not found"),
))
}
};
match self.rpc_handler.new_onchain_address().await {
Ok(address) if !address.is_empty() => {
crate::content_invoice::record_pending(&address, content_id, price_sats).await;
let body = serde_json::json!({
"address": address,
"amount_sats": price_sats,
});
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
_ => {
let body = serde_json::json!({
"error": "Could not generate an on-chain address (is the wallet ready?)"
});
Ok(build_response(
StatusCode::SERVICE_UNAVAILABLE,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
}
}
/// Seller side (#46): report whether an on-chain payment to a previously-
/// issued address has arrived (>= price, >= 1 conf). Path:
/// GET /content/{id}/onchain-status/{address}. Marks the entitlement paid.
pub(super) async fn handle_content_onchain_status(
&self,
path: &str,
) -> Result<Response<hyper::Body>> {
let rest = path.strip_prefix("/content/").unwrap_or("");
let (content_id, address) = match rest.split_once("/onchain-status/") {
Some((id, addr)) => (id, addr),
None => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid request"),
))
}
};
if content_id.is_empty() || !is_valid_app_id(content_id) || address.is_empty() {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid request"),
));
}
// The address must be one we issued for exactly this content item.
let price = match crate::content_invoice::lookup(address).await {
Some((cid, price)) if cid == content_id => price,
_ => {
return Ok(build_response(
StatusCode::NOT_FOUND,
"application/json",
hyper::Body::from(r#"{"error":"Unknown address"}"#),
))
}
};
let mut paid = crate::content_invoice::is_paid_for(address, content_id).await;
if !paid {
if let Ok(true) = self.rpc_handler.onchain_received(address, price).await {
crate::content_invoice::mark_paid(address).await;
paid = true;
}
}
let body = serde_json::json!({ "paid": paid });
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(serde_json::to_vec(&body).unwrap_or_default()),
))
}
/// Serve a degraded preview of paid content (blurred image or first 2% of video).
pub(super) async fn handle_content_preview(
path: &str,
config: &Config,
) -> Result<Response<hyper::Body>> {
// Path format: /content/{id}/preview
let content_id = path
.strip_prefix("/content/")
.and_then(|s| s.strip_suffix("/preview"))
.unwrap_or("");
if content_id.is_empty() || !is_valid_app_id(content_id) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"text/plain",
hyper::Body::from("Invalid content ID"),
));
}
match content_server::serve_content_preview(&config.data_dir, content_id).await {
Ok(content_server::PreviewResult::FullContent(bytes, mime_type)) => {
let len = bytes.len();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", mime_type)
.header("Content-Length", len.to_string())
.body(hyper::Body::from(bytes))
.unwrap())
}
Ok(content_server::PreviewResult::BlurPreview(bytes, mime_type)) => {
let len = bytes.len();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", mime_type)
.header("Content-Length", len.to_string())
.header("X-Content-Preview", "blur")
.body(hyper::Body::from(bytes))
.unwrap())
}
Ok(content_server::PreviewResult::TruncatedPreview(bytes, mime_type, total_size)) => {
let len = bytes.len();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", mime_type)
.header("Content-Length", len.to_string())
.header("X-Content-Preview", "truncated")
.header("X-Content-Total-Size", total_size.to_string())
.body(hyper::Body::from(bytes))
.unwrap())
}
Ok(content_server::PreviewResult::PreviewUnavailable) => Ok(Response::builder()
.status(StatusCode::UNSUPPORTED_MEDIA_TYPE)
.header("Content-Type", "text/plain")
.header("X-Content-Preview", "unavailable")
.body(hyper::Body::from(
"Preview unavailable for this media (needs re-encoding)",
))
.unwrap()),
Ok(content_server::PreviewResult::NotFound) | Err(_) => Ok(build_response(
StatusCode::NOT_FOUND,
"text/plain",
hyper::Body::from("Preview not available"),
)),
}
}
}
core/archipelago/src/api/handler/dwn.rs
+201
View File
@@ -0,0 +1,201 @@
use super::build_response;
use crate::config::Config;
use crate::network::dwn_store::DwnStore;
use anyhow::Result;
use hyper::{Response, StatusCode};
use super::ApiHandler;
impl ApiHandler {
/// DWN health endpoint — returns store stats.
pub(super) async fn handle_dwn_health(config: &Config) -> Result<Response<hyper::Body>> {
match DwnStore::new(&config.data_dir).await {
Ok(store) => {
let stats = store
.stats()
.await
.unwrap_or(crate::network::dwn_store::StoreStats {
message_count: 0,
protocol_count: 0,
total_bytes: 0,
});
let body = serde_json::json!({
"status": "ok",
"message_count": stats.message_count,
"protocol_count": stats.protocol_count,
"total_bytes": stats.total_bytes,
});
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(hyper::Body::from(body.to_string()))
.unwrap())
}
Err(_) => Ok(build_response(
StatusCode::SERVICE_UNAVAILABLE,
"application/json",
hyper::Body::from(r#"{"status":"unavailable"}"#),
)),
}
}
/// DWN message processing endpoint — handles RecordsWrite, RecordsQuery, RecordsRead, RecordsDelete.
/// Supports batch processing: all messages in the array are processed.
pub(super) async fn handle_dwn_message(
body: hyper::body::Bytes,
config: &Config,
) -> Result<Response<hyper::Body>> {
let request: serde_json::Value = match serde_json::from_slice(&body) {
Ok(v) => v,
Err(e) => {
let err = serde_json::json!({"error": format!("Invalid JSON: {}", e)});
return Ok(Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("Content-Type", "application/json")
.body(hyper::Body::from(err.to_string()))
.unwrap());
}
};
// Collect all messages to process
let messages: Vec<serde_json::Value> = if request.get("message").is_some() {
vec![request["message"].clone()]
} else if let Some(msgs) = request["messages"].as_array() {
msgs.clone()
} else {
vec![serde_json::Value::Null]
};
let store = DwnStore::new(&config.data_dir).await?;
let mut results = Vec::new();
for message in &messages {
let interface = message["descriptor"]["interface"].as_str().unwrap_or("");
let method = message["descriptor"]["method"].as_str().unwrap_or("");
let result = match (interface, method) {
("Records", "Write") => {
let author = message["author"].as_str().unwrap_or("unknown");
let protocol = message["descriptor"]["protocol"].as_str();
let schema = message["descriptor"]["schema"].as_str();
let data_format = message["descriptor"]["dataFormat"].as_str();
let data = message.get("data").cloned();
// Deduplicate: check if recordId already exists
if let Some(record_id) = message["recordId"].as_str() {
if store.read_message(record_id).await.ok().flatten().is_some() {
serde_json::json!({"status": {"code": 200, "detail": "Already exists"}})
} else {
match store
.write_message(author, protocol, schema, data_format, data)
.await
{
Ok(msg) => {
serde_json::json!({"status": {"code": 202}, "entry": msg})
}
Err(e) => {
serde_json::json!({"status": {"code": 500, "detail": e.to_string()}})
}
}
}
} else {
match store
.write_message(author, protocol, schema, data_format, data)
.await
{
Ok(msg) => serde_json::json!({"status": {"code": 202}, "entry": msg}),
Err(e) => {
serde_json::json!({"status": {"code": 500, "detail": e.to_string()}})
}
}
}
}
("Records", "Query") => {
let query = crate::network::dwn_store::MessageQuery {
protocol: message["descriptor"]["filter"]["protocol"]
.as_str()
.map(|s| s.to_string()),
schema: message["descriptor"]["filter"]["schema"]
.as_str()
.map(|s| s.to_string()),
author: message["descriptor"]["filter"]["author"]
.as_str()
.map(|s| s.to_string()),
date_from: message["descriptor"]["filter"]["dateFrom"]
.as_str()
.map(|s| s.to_string()),
date_to: message["descriptor"]["filter"]["dateTo"]
.as_str()
.map(|s| s.to_string()),
limit: message["descriptor"]["filter"]["limit"]
.as_u64()
.map(|n| n as usize),
};
match store.query_messages(&query).await {
Ok(messages) => {
serde_json::json!({"status": {"code": 200}, "entries": messages})
}
Err(e) => {
serde_json::json!({"status": {"code": 500, "detail": e.to_string()}})
}
}
}
("Records", "Read") => {
let record_id = message["descriptor"]["recordId"].as_str().unwrap_or("");
match store.read_message(record_id).await {
Ok(Some(msg)) => {
serde_json::json!({"status": {"code": 200}, "entry": msg})
}
Ok(None) => {
serde_json::json!({"status": {"code": 404, "detail": "Record not found"}})
}
Err(e) => {
serde_json::json!({"status": {"code": 500, "detail": e.to_string()}})
}
}
}
("Records", "Delete") => {
let record_id = message["descriptor"]["recordId"].as_str().unwrap_or("");
match store.delete_message(record_id).await {
Ok(true) => serde_json::json!({"status": {"code": 200}}),
Ok(false) => {
serde_json::json!({"status": {"code": 404, "detail": "Record not found"}})
}
Err(e) => {
serde_json::json!({"status": {"code": 500, "detail": e.to_string()}})
}
}
}
_ => {
serde_json::json!({"status": {"code": 400, "detail": format!("Unknown method: {}.{}", interface, method)}})
}
};
results.push(result);
}
// Return single result for single message, array for batch
let (response_body, http_status) = if results.len() == 1 {
let result = &results[0];
let status_code = result["status"]["code"].as_u64().unwrap_or(200);
let http_status = match status_code {
202 => StatusCode::ACCEPTED,
400 => StatusCode::BAD_REQUEST,
404 => StatusCode::NOT_FOUND,
500 => StatusCode::INTERNAL_SERVER_ERROR,
_ => StatusCode::OK,
};
(result.to_string(), http_status)
} else {
(
serde_json::json!({"replies": results}).to_string(),
StatusCode::OK,
)
};
Ok(build_response(
http_status,
"application/json",
hyper::Body::from(response_body),
))
}
}
core/archipelago/src/api/handler/mod.rs
+619
View File
@@ -0,0 +1,619 @@
mod blob;
mod content;
mod dwn;
mod node_message;
mod proxy;
mod remote_input;
mod remote_relay;
mod websocket;
use crate::api::rpc::RpcHandler;
use crate::blobs::BlobStore;
use crate::config::Config;
use crate::container::{ContainerOrchestrator, DevContainerOrchestrator};
use crate::monitoring::MetricsStore;
use crate::session::{self, SessionStore};
use crate::state::StateManager;
use anyhow::Result;
use hyper::{Method, Request, Response, StatusCode};
use sha2::{Digest, Sha256};
use std::sync::Arc;
use tokio::sync::broadcast;
use tracing::debug;
/// Build an HTTP response without unwrap. Falls back to a plain 500 if builder fails.
// Used by handler submodules after unwrap elimination
#[allow(dead_code)]
pub(super) fn build_response(
status: StatusCode,
content_type: &str,
body: hyper::Body,
) -> Response<hyper::Body> {
Response::builder()
.status(status)
.header("Content-Type", content_type)
.body(body)
.unwrap_or_else(|_| Response::new(hyper::Body::from("Internal error")))
}
pub struct ApiHandler {
config: Config,
rpc_handler: Arc<RpcHandler>,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
session_store: SessionStore,
/// Broadcast channel for relaying companion app input to remote browsers.
input_relay_tx: broadcast::Sender<String>,
/// Reverse broadcast channel: the kiosk browser publishes "open this URL
/// externally" requests here, and the companion (phone) socket forwards them
/// to the phone's default browser. Lets "open in external browser" apps —
/// which the kiosk can't usefully open itself — launch on the controller.
external_open_tx: broadcast::Sender<String>,
/// Content-addressed blob store for attachments shared over mesh/federation.
blob_store: Arc<BlobStore>,
/// Our own node pubkey (hex) — used to self-sign debug/test capabilities.
self_pubkey_hex: String,
}
impl ApiHandler {
pub async fn new(
config: Config,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
orchestrator: Option<Arc<dyn ContainerOrchestrator>>,
dev_orchestrator: Option<Arc<DevContainerOrchestrator>>,
) -> Result<Self> {
let session_store = SessionStore::new().await;
let rpc_handler = Arc::new(
RpcHandler::new(
config.clone(),
state_manager.clone(),
metrics_store.clone(),
session_store.clone(),
orchestrator,
dev_orchestrator,
)
.await?,
);
let (input_relay_tx, _) = broadcast::channel(64);
let (external_open_tx, _) = broadcast::channel(16);
// Derive a blob-store capability key from the node's Ed25519 signing
// key. SHA-256 domain-separated so rotating the identity rotates
// every outstanding capability token (intentional — prevents a
// replaced node from honouring old caps).
let identity_dir = config.data_dir.join("identity");
let identity = crate::identity::NodeIdentity::load_or_create(&identity_dir).await?;
let mut hasher = Sha256::new();
hasher.update(identity.signing_key().to_bytes());
hasher.update(b"|archipelago-blob-cap-v1");
let mut cap_key = [0u8; 32];
cap_key.copy_from_slice(&hasher.finalize());
let blob_store = Arc::new(BlobStore::open(&config.data_dir, cap_key).await?);
let self_pubkey_hex = hex::encode(identity.signing_key().verifying_key().as_bytes());
// Share blob store with the RPC layer so mesh.send-content /
// mesh.fetch-content can reach the same instance (single cap_key,
// single on-disk root) without re-opening it.
rpc_handler
.set_blob_store(blob_store.clone(), self_pubkey_hex.clone())
.await;
Ok(Self {
config,
rpc_handler,
state_manager,
metrics_store,
session_store,
input_relay_tx,
external_open_tx,
blob_store,
self_pubkey_hex,
})
}
/// Access the RPC handler (for service initialization after construction).
pub fn rpc_handler(&self) -> &Arc<RpcHandler> {
&self.rpc_handler
}
/// Check if the request has a valid session cookie.
async fn is_authenticated(&self, headers: &hyper::HeaderMap) -> bool {
match session::extract_session_cookie(headers) {
Some(token) => self.session_store.validate(&token).await,
None => false,
}
}
/// Server-side fetch of the upstream app catalog so the browser can
/// load it without fighting CORS (upstream Gitea emits no ACAO) or
/// CSP (the fallback IP-port URL isn't in `connect-src`). The upstream
/// list is derived from the operator's configured container registries
/// so switching mirrors in Settings changes the App Store source too —
/// each active registry contributes one Gitea `raw/branch/main/catalog.json`
/// URL (http or https per `tls_verify`), tried in priority order.
/// If registry config can't be loaded, falls back to the hardcoded OVH
/// URL so the App Store still renders on nodes that haven't persisted
/// a registry config yet. 15s total timeout.
async fn handle_app_catalog_proxy(&self) -> Result<Response<hyper::Body>> {
let mut upstreams: Vec<String> = Vec::new();
if let Ok(config) = crate::container::registry::load_registries(&self.config.data_dir).await
{
for reg in config.active_registries() {
let scheme = if reg.tls_verify { "https" } else { "http" };
// Gitea raw URL: <scheme>://<host>/<namespace>/app-catalog/raw/branch/main/catalog.json.
// reg.url already includes the namespace (e.g. "host/lfg2025"),
// so we just tack on the repo + raw path.
upstreams.push(format!(
"{}://{}/app-catalog/raw/branch/main/catalog.json",
scheme, reg.url
));
}
}
if upstreams.is_empty() {
upstreams.push(
"http://146.59.87.168:3000/lfg2025/app-catalog/raw/branch/main/catalog.json"
.to_string(),
);
}
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(15))
.build()
{
Ok(c) => c,
Err(e) => {
return Ok(build_response(
hyper::StatusCode::INTERNAL_SERVER_ERROR,
"text/plain",
hyper::Body::from(format!("client build failed: {}", e)),
));
}
};
for url in &upstreams {
match client.get(url).send().await {
Ok(resp) if resp.status().is_success() => {
if let Ok(bytes) = resp.bytes().await {
return Ok(Response::builder()
.status(hyper::StatusCode::OK)
.header("Content-Type", "application/json")
.header("Cache-Control", "public, max-age=3600")
.body(hyper::Body::from(bytes))
.unwrap_or_else(|_| {
Response::new(hyper::Body::from("proxy response build failed"))
}));
}
}
_ => continue,
}
}
Ok(build_response(
hyper::StatusCode::BAD_GATEWAY,
"text/plain",
hyper::Body::from("all upstream catalog URLs failed"),
))
}
/// Build a 401 Unauthorized JSON response.
fn unauthorized() -> Response<hyper::Body> {
let body = serde_json::json!({ "error": "Unauthorized" });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header("Content-Type", "application/json")
.body(hyper::Body::from(body_bytes))
.unwrap()
}
/// A 401 that still carries CORS headers, for endpoints fetched
/// cross-origin by same-node app UIs (e.g. the LND wallet UI on its own
/// port). Without the ACAO header the browser surfaces an opaque CORS
/// error instead of the 401, so the app can't tell it just needs auth.
/// `origin` is the already-validated reflect value from `app_cors_origin`
/// (empty string when the origin isn't allowed → no CORS header added).
fn unauthorized_cors(origin: &str) -> Response<hyper::Body> {
let body = serde_json::json!({ "error": "Unauthorized" });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
let mut builder = Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header("Content-Type", "application/json")
.header("Vary", "Origin");
if !origin.is_empty() {
builder = builder
.header("Access-Control-Allow-Origin", origin)
.header("Access-Control-Allow-Credentials", "true");
}
builder.body(hyper::Body::from(body_bytes)).unwrap()
}
/// Allowed CORS origins derived from the config host IP.
fn allowed_origins(&self) -> Vec<String> {
let mut origins = vec![
format!("http://{}", self.config.host_ip),
format!("https://{}", self.config.host_ip),
];
if self.config.dev_mode {
origins.push("http://localhost:8100".to_string()); // Vite dev server
}
origins
}
/// Validate the Origin header against allowed origins.
/// Returns the matched origin if valid, None if cross-origin is not allowed.
fn validate_origin(&self, headers: &hyper::HeaderMap) -> Option<String> {
let origin = headers.get("origin").and_then(|v| v.to_str().ok())?;
let allowed = self.allowed_origins();
if allowed.iter().any(|a| a == origin) {
Some(origin.to_string())
} else {
None
}
}
/// Permissive origin check for the share-to-mesh iframe intent: any scheme
/// http(s):// followed by the configured host_ip, optionally `:port`. Apps
/// proxied under other ports (APP_PORTS) call this from within the same
/// node, so they share host_ip but not port. The session cookie still has
/// to be valid — this is a sanity check, not the primary auth.
fn validate_app_origin(&self, headers: &hyper::HeaderMap) -> Option<String> {
let origin = headers.get("origin").and_then(|v| v.to_str().ok())?;
// Allow localhost dev server too so the Vite frontend can exercise it.
if self.config.dev_mode && origin == "http://localhost:8100" {
return Some(origin.to_string());
}
let host_ip = &self.config.host_ip;
let matches = |scheme: &str| -> bool {
let prefix = format!("{}{}", scheme, host_ip);
if origin == prefix {
return true;
}
let with_port = format!("{}:", prefix);
origin.starts_with(&with_port)
&& origin[with_port.len()..]
.bytes()
.all(|b| b.is_ascii_digit())
};
if matches("http://") || matches("https://") {
Some(origin.to_string())
} else {
None
}
}
/// CORS origin to echo for same-node app → backend calls (e.g. the LND
/// wallet UI, served on its own APP_PORTS port). Such apps share the node's
/// host but use a different port, so the strict allowlist (`host_ip`, no
/// port) rejects them and the browser gets no `Access-Control-Allow-Origin`
/// header ("blocked by CORS policy"). Reflect the Origin when its host
/// matches the request's own `Host` header — i.e. the app lives on the same
/// address the node is being reached by, which transparently covers the LAN
/// IP, the Tailscale IP, localhost, and the `.onion` address without needing
/// to enumerate them. Auth is still enforced by the session cookie; this
/// only authorizes the browser to *read* the reply. Returns "" (no echoed
/// origin) when there is no match.
fn app_cors_origin(&self, headers: &hyper::HeaderMap) -> String {
if let Some(origin) = self.validate_origin(headers) {
return origin;
}
let Some(origin) = headers.get("origin").and_then(|v| v.to_str().ok()) else {
return String::new();
};
// host portion (no scheme, no port) of an `scheme://host[:port]` value
let host_of = |s: &str| -> Option<String> {
let after_scheme = s.split_once("://").map(|(_, r)| r).unwrap_or(s);
let host_port = after_scheme.split('/').next().unwrap_or(after_scheme);
let host = host_port
.rsplit_once(':')
.map(|(h, _)| h)
.unwrap_or(host_port);
(!host.is_empty()).then(|| host.to_string())
};
let origin_host = host_of(origin);
let req_host = headers
.get(hyper::header::HOST)
.and_then(|v| v.to_str().ok())
.and_then(host_of);
match (origin_host, req_host) {
(Some(o), Some(r)) if o == r => origin.to_string(),
_ => String::new(),
}
}
pub async fn handle_request(&self, req: Request<hyper::Body>) -> Result<Response<hyper::Body>> {
let path = req.uri().path().to_string();
let method = req.method().clone();
// Handle CORS preflight for all routes
if method == Method::OPTIONS {
let mut builder = Response::builder()
.status(StatusCode::NO_CONTENT)
.header("Vary", "Origin");
let preflight_origin = self.app_cors_origin(req.headers());
if !preflight_origin.is_empty() {
builder = builder
.header("Access-Control-Allow-Origin", &preflight_origin)
.header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
.header("Access-Control-Allow-Headers", "Content-Type, X-CSRF-Token")
.header("Access-Control-Allow-Credentials", "true");
}
return Ok(builder.body(hyper::Body::empty()).unwrap());
}
// WebSocket upgrade — validate session before upgrading
if method == Method::GET && path == "/ws/db" {
if !self.is_authenticated(req.headers()).await {
tracing::warn!("401 WebSocket /ws/db — session invalid or missing");
return Ok(Self::unauthorized());
}
return Self::handle_websocket(
req,
self.state_manager.clone(),
self.metrics_store.clone(),
)
.await;
}
// Remote input WebSocket — companion app sends keyboard/mouse events
if method == Method::GET && path == "/ws/remote-input" {
if !self.is_authenticated(req.headers()).await {
tracing::warn!("401 WebSocket /ws/remote-input — session invalid or missing");
return Ok(Self::unauthorized());
}
return Self::handle_remote_input(
req,
self.input_relay_tx.clone(),
self.external_open_tx.subscribe(),
)
.await;
}
// Remote relay WebSocket — browser receives companion input events
if method == Method::GET && path == "/ws/remote-relay" {
if !self.is_authenticated(req.headers()).await {
tracing::warn!("401 WebSocket /ws/remote-relay — session invalid or missing");
return Ok(Self::unauthorized());
}
return Self::handle_remote_relay(
req,
self.input_relay_tx.subscribe(),
self.external_open_tx.clone(),
)
.await;
}
// Convert body to bytes for non-WS routes
let headers = req.headers().clone();
let query_string = req.uri().query().map(|s| s.to_string()).unwrap_or_default();
let (parts, body) = req.into_parts();
let body_bytes = hyper::body::to_bytes(body)
.await
.map_err(|e| anyhow::anyhow!("Failed to read body: {}", e))?;
let req_with_bytes = Request::from_parts(parts, hyper::Body::from(body_bytes.clone()));
debug!("{} {}", method, path);
match (method, path.as_str()) {
// RPC — auth is handled inside rpc handler per-method
(Method::POST, "/rpc/v1") => self.rpc_handler.clone().handle(req_with_bytes).await,
// Health — unauthenticated, returns JSON with service status
(Method::GET, "/health") => {
let recovery_complete = crate::crash_recovery::is_recovery_complete();
let uptime = crate::crash_recovery::uptime_seconds();
let health_status = if recovery_complete { "ok" } else { "degraded" };
let status = serde_json::json!({
"status": health_status,
"crash_recovery_complete": recovery_complete,
"uptime_seconds": uptime,
"version": env!("CARGO_PKG_VERSION"),
"services": {
"rpc": true,
"sessions": true,
}
});
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(hyper::Body::from(
serde_json::to_vec(&status).unwrap_or_default(),
))
.unwrap())
}
// Node message — P2P endpoint (authenticated by source validation, not cookie)
(Method::POST, "/archipelago/node-message") => {
Self::handle_node_message(body_bytes).await
}
// Mesh typed envelope relay over federation — peers POST
// pre-encoded TypedEnvelope wire bytes here when the envelope is
// too large for a single LoRa frame (primarily ContentRef). No
// session auth: the body carries a pubkey + ed25519 signature
// over the wire bytes which we verify before dispatching.
(Method::POST, "/archipelago/mesh-typed") => {
Self::handle_mesh_typed_relay(self.rpc_handler.clone(), body_bytes).await
}
// Blob upload — local/session use only. Session-authenticated so
// only the node owner can push attachments into the blob store.
(Method::POST, "/api/blob") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
Self::handle_blob_upload(
&self.blob_store,
&self.self_pubkey_hex,
&self.config.data_dir,
&headers,
body_bytes,
)
.await
}
// Share-to-mesh intent — marketplace app iframes POST a file here
// to stage it as a mesh attachment. Same body format as /api/blob
// (raw bytes + X-Blob-Mime/X-Blob-Filename headers). The app is
// expected to postMessage `{type:'share-to-mesh', cid, ...}` to
// its parent window afterwards so the Mesh view can pick it up.
// Authenticated by session cookie + a relaxed Origin check (any
// port on the archipelago host is allowed, so proxied apps on
// their own ports can reach it with credentials:'include').
(Method::POST, "/api/share-to-mesh") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
let origin = match self.validate_app_origin(&headers) {
Some(o) => o,
None => {
return Ok(build_response(
StatusCode::FORBIDDEN,
"text/plain",
hyper::Body::from("origin not allowed"),
))
}
};
Self::handle_share_to_mesh(
&self.blob_store,
&self.self_pubkey_hex,
&headers,
body_bytes,
&origin,
)
.await
}
// Blob download — peer-facing. No session required; authenticated
// by HMAC capability token signed when the blob ref was shared.
(Method::GET, p) if p.starts_with("/blob/") => {
Self::handle_blob_download(&self.blob_store, p, &query_string).await
}
// Content preview — degraded previews for paid content (no auth, no payment)
(Method::GET, p) if p.starts_with("/content/") && p.ends_with("/preview") => {
Self::handle_content_preview(p, &self.config).await
}
// Lightning-invoice peer-file sale (#46): mint invoice / poll settlement
(Method::GET, p) if p.starts_with("/content/") && p.ends_with("/invoice") => {
self.handle_content_invoice(p).await
}
(Method::GET, p) if p.starts_with("/content/") && p.contains("/invoice-status/") => {
self.handle_content_invoice_status(p).await
}
// On-chain peer-file sale (#46): issue address / poll for payment
(Method::GET, p) if p.starts_with("/content/") && p.contains("/onchain-status/") => {
self.handle_content_onchain_status(p).await
}
(Method::GET, p) if p.starts_with("/content/") && p.ends_with("/onchain") => {
self.handle_content_onchain(p).await
}
// Content serving — peers access shared content over Tor (no session auth)
(Method::GET, p) if p.starts_with("/content/") => {
Self::handle_content_request(p, &headers, &self.config).await
}
// Content catalog — list available content (no session auth, for peers)
(Method::GET, "/content") => Self::handle_content_catalog(&self.config).await,
// Electrs status — unauthenticated (read-only sync status)
(Method::GET, "/electrs-status") => Self::handle_electrs_status().await,
(Method::GET, "/bitcoin-status") => Self::handle_bitcoin_status().await,
// App-catalog proxy — fetches catalog.json from the configured
// upstream URLs server-side so the browser doesn't hit CORS
// (upstream Gitea has no ACAO header) or CSP (IP-port upstream
// falls outside `connect-src`). Session-authenticated so only
// the logged-in node owner can spin up fetches.
(Method::GET, "/api/app-catalog") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
self.handle_app_catalog_proxy().await
}
// LND connect info — nginx validates session cookie (presence check),
// backend is bound to 127.0.0.1 so only nginx can reach it.
// No backend auth check here because the LND UI iframe fetches this
// endpoint and the session cookie flow is validated at the nginx layer.
(Method::GET, "/lnd-connect-info") => {
let origin = self.app_cors_origin(&headers);
Self::handle_lnd_connect_info(self.rpc_handler.clone(), &origin).await
}
// Container logs — requires session
(Method::GET, path) if path.starts_with("/api/container/logs") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
let origin = self.validate_origin(&headers).unwrap_or_default();
Self::handle_container_logs_http(self.rpc_handler.clone(), path, &origin).await
}
// Peer content streaming proxy — Range-streams a peer's media file
// so <video>/<audio> can seek/play (B3). Same-origin, session-gated.
(Method::GET, p) if p.starts_with("/api/peer-content/") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
self.handle_peer_content_stream(p, &headers).await
}
// LND proxy — requires session. The LND wallet UI calls this
// cross-origin from its own app port, so even the 401 must carry
// CORS headers; otherwise the browser reports a bare CORS failure
// ("No 'Access-Control-Allow-Origin' header") instead of a
// readable 401 the UI can act on.
(Method::GET, path) if path.starts_with("/proxy/lnd/") => {
let origin = self.app_cors_origin(&headers);
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized_cors(&origin));
}
Self::handle_lnd_proxy(self.rpc_handler.clone(), path, &origin).await
}
// DWN health — unauthenticated
(Method::GET, "/dwn/health") => Self::handle_dwn_health(&self.config).await,
// DWN message processing — peers access over Tor for sync (no session auth)
(Method::POST, "/dwn") => Self::handle_dwn_message(body_bytes, &self.config).await,
_ => Ok(Response::builder()
.status(StatusCode::NOT_FOUND)
.body(hyper::Body::from("Not Found"))
.unwrap()),
}
}
}
/// Validate that an app ID matches the safe pattern: lowercase alphanumeric + hyphens.
fn is_valid_app_id(id: &str) -> bool {
!id.is_empty()
&& id.len() <= 64
&& id
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
&& id.as_bytes()[0] != b'-'
}
/// Validate that a pubkey is a 64-char hex string.
fn is_valid_pubkey_hex(s: &str) -> bool {
s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit())
}
/// Strip newlines and ANSI escape sequences from strings before logging.
fn sanitize_log_string(s: &str) -> String {
s.replace('\n', "\\n")
.replace('\r', "\\r")
.replace('\x1b', "")
}
/// Strip HTML-sensitive characters to prevent XSS when stored/rendered.
fn sanitize_html(s: &str) -> String {
s.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&#x27;")
}
core/archipelago/src/api/handler/node_message.rs
+265
View File
@@ -0,0 +1,265 @@
use super::build_response;
use crate::api::rpc::RpcHandler;
use crate::node_message as node_msg;
use anyhow::Result;
use hyper::{Response, StatusCode};
use std::sync::Arc;
use super::{is_valid_pubkey_hex, sanitize_html, sanitize_log_string, ApiHandler};
impl ApiHandler {
pub(super) async fn handle_node_message(
body: hyper::body::Bytes,
) -> Result<Response<hyper::Body>> {
#[derive(serde::Deserialize)]
struct Incoming {
from_pubkey: Option<String>,
from_name: Option<String>,
message: Option<String>,
signature: Option<String>,
#[serde(default)]
encrypted: bool,
#[serde(default)]
msg_id: Option<String>,
}
let incoming: Incoming = serde_json::from_slice(&body).unwrap_or(Incoming {
from_pubkey: None,
from_name: None,
message: None,
signature: None,
encrypted: false,
msg_id: None,
});
if let (Some(from), Some(msg)) = (incoming.from_pubkey.as_ref(), incoming.message.as_ref())
{
// Validate from_pubkey is a valid hex ed25519 pubkey
if !is_valid_pubkey_hex(from) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"Invalid pubkey format"}"#),
));
}
// Verify ed25519 signature if provided (required for trusted messages)
if let Some(sig_hex) = &incoming.signature {
match crate::identity::NodeIdentity::verify(from, msg.as_bytes(), sig_hex) {
Ok(true) => {}
_ => {
return Ok(build_response(
StatusCode::FORBIDDEN,
"application/json",
hyper::Body::from(r#"{"error":"Invalid signature"}"#),
));
}
}
}
// Decrypt if the message is E2E encrypted
let plaintext = if incoming.encrypted {
// Load our identity to derive shared secret
let data_dir = std::path::Path::new("/var/lib/archipelago");
let identity_dir = data_dir.join("identity");
match crate::identity::NodeIdentity::load_or_create(&identity_dir).await {
Ok(node_id) => {
match node_msg::decrypt_from_peer(node_id.signing_key(), from, msg) {
Ok(decrypted) => {
tracing::info!(
"Decrypted E2E message from {}...",
&from[..16.min(from.len())]
);
decrypted
}
Err(e) => {
tracing::warn!(
"E2E decryption failed from {}: {}",
&from[..16.min(from.len())],
e
);
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"Decryption failed"}"#),
));
}
}
}
Err(e) => {
tracing::warn!("Cannot decrypt: identity load failed: {}", e);
msg.clone()
}
}
} else {
msg.clone()
};
// Detect a `connection_accepted` reply: the remote peer just
// approved an outbound request we sent, so mirror their add on
// our side (bidirectional peering without a manual second
// click). JSON-shape only — any non-matching payload stays in
// the normal received-messages store below.
if let Ok(val) = serde_json::from_str::<serde_json::Value>(&plaintext) {
if val.get("type").and_then(|v| v.as_str()) == Some("connection_accepted") {
if let (Some(their_onion), Some(their_pubkey)) = (
val.get("from_onion").and_then(|v| v.as_str()),
val.get("from_pubkey").and_then(|v| v.as_str()),
) {
let data_dir = std::path::Path::new("/var/lib/archipelago");
let peer = crate::peers::KnownPeer {
onion: their_onion.to_string(),
pubkey: their_pubkey.to_string(),
name: val
.get("from_name")
.and_then(|v| v.as_str())
.map(String::from),
added_at: Some(chrono::Utc::now().to_rfc3339()),
};
match crate::peers::add_peer(data_dir, peer).await {
Ok(_) => tracing::info!(
from = %sanitize_log_string(from),
"Auto-added peer after connection_accepted"
),
Err(e) => tracing::warn!(
from = %sanitize_log_string(from),
error = %e,
"Failed to auto-add peer on connection_accepted"
),
}
}
return Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(r#"{"ok":true,"handled":"connection_accepted"}"#),
));
}
if let Some(handled) =
crate::api::rpc::bitcoin_relay::record_incoming_relay_message(
std::path::Path::new("/var/lib/archipelago"),
from,
incoming.from_name.as_deref(),
&val,
)
.await?
{
return Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(format!(r#"{{"ok":true,"handled":"{}"}}"#, handled)),
));
}
}
let safe_from = sanitize_log_string(from);
let safe_msg = sanitize_log_string(&plaintext);
tracing::info!("Received message from {}: {}", safe_from, safe_msg);
let clean_from = sanitize_html(from);
let clean_msg = sanitize_html(&plaintext);
let clean_name = incoming.from_name.as_deref().map(sanitize_html);
node_msg::store_received(
&clean_from,
&clean_msg,
clean_name.as_deref(),
incoming.msg_id.as_deref(),
)
.await;
}
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(r#"{"ok":true}"#),
))
}
/// Federation-routed mesh typed envelope. Body:
/// `{from_pubkey, from_name?, typed_envelope_b64, signature}`
/// Signature is ed25519 over the raw wire bytes, verified against
/// from_pubkey before dispatch.
pub(super) async fn handle_mesh_typed_relay(
rpc_handler: Arc<RpcHandler>,
body: hyper::body::Bytes,
) -> Result<Response<hyper::Body>> {
use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
#[derive(serde::Deserialize)]
struct Incoming {
from_pubkey: String,
#[serde(default)]
from_name: Option<String>,
typed_envelope_b64: String,
signature: String,
}
let incoming: Incoming = match serde_json::from_slice(&body) {
Ok(v) => v,
Err(e) => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(format!(r#"{{"error":"bad json: {}"}}"#, e)),
));
}
};
if !is_valid_pubkey_hex(&incoming.from_pubkey) {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"invalid pubkey"}"#),
));
}
let wire = match BASE64.decode(incoming.typed_envelope_b64.as_bytes()) {
Ok(v) => v,
Err(_) => {
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(r#"{"error":"bad base64"}"#),
));
}
};
match crate::identity::NodeIdentity::verify(
&incoming.from_pubkey,
&wire,
&incoming.signature,
) {
Ok(true) => {}
_ => {
return Ok(build_response(
StatusCode::FORBIDDEN,
"application/json",
hyper::Body::from(r#"{"error":"signature rejected"}"#),
));
}
}
// Inject into mesh state via the shared MeshService. Mirrors a radio
// receive, so the message lands in the same chat stream as LoRa-
// delivered messages from the same peer.
let service = rpc_handler.mesh_service_arc();
let svc_guard = service.read().await;
let Some(svc) = svc_guard.as_ref() else {
return Ok(build_response(
StatusCode::SERVICE_UNAVAILABLE,
"application/json",
hyper::Body::from(r#"{"error":"mesh not running"}"#),
));
};
if let Err(e) = svc
.inject_typed_from_federation(
&incoming.from_pubkey,
incoming.from_name.as_deref(),
wire,
)
.await
{
tracing::warn!("mesh-typed relay inject failed: {}", e);
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(format!(r#"{{"error":"{}"}}"#, e)),
));
}
Ok(build_response(
StatusCode::OK,
"application/json",
hyper::Body::from(r#"{"ok":true}"#),
))
}
}
core/archipelago/src/api/handler/proxy.rs
+268
View File
@@ -0,0 +1,268 @@
use super::build_response;
use crate::api::rpc::lnd::LND_REST_BASE_URL;
use crate::api::rpc::RpcHandler;
use crate::bitcoin_status;
use crate::electrs_status;
use anyhow::Result;
use hyper::{Response, StatusCode};
use std::sync::Arc;
use super::{is_valid_app_id, ApiHandler};
impl ApiHandler {
pub(super) async fn handle_container_logs_http(
rpc: Arc<RpcHandler>,
path: &str,
cors_origin: &str,
) -> Result<Response<hyper::Body>> {
let query = path
.strip_prefix("/api/container/logs")
.and_then(|s| s.strip_prefix('?'))
.unwrap_or("");
let params: std::collections::HashMap<String, String> = query
.split('&')
.filter_map(|p| {
let mut it = p.splitn(2, '=');
let k = it.next()?.to_string();
let v = it.next()?.to_string();
Some((k, v))
})
.collect();
let app_id = params.get("app_id").map(|s| s.as_str()).unwrap_or("lnd");
// Validate app_id format
if !is_valid_app_id(app_id) {
let body = serde_json::json!({ "error": "Invalid app_id" });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
return Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(body_bytes),
));
}
let lines = params
.get("lines")
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(200);
match rpc.get_container_logs_value(app_id, lines).await {
Ok(value) => {
let body = serde_json::json!({ "result": value });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.header("Access-Control-Allow-Origin", cors_origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
.body(hyper::Body::from(body_bytes))
.unwrap())
}
Err(e) => {
let body = serde_json::json!({ "error": e.to_string() });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.header("Content-Type", "application/json")
.header("Access-Control-Allow-Origin", cors_origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
.body(hyper::Body::from(body_bytes))
.unwrap())
}
}
}
pub(super) async fn handle_electrs_status() -> Result<Response<hyper::Body>> {
let status = electrs_status::get_electrs_sync_status().await;
let body = serde_json::to_vec(&status).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.header("Cache-Control", "no-store")
.body(hyper::Body::from(body))
.unwrap_or_else(|_| Response::new(hyper::Body::from("{}"))))
}
pub(super) async fn handle_bitcoin_status() -> Result<Response<hyper::Body>> {
let status = bitcoin_status::get_bitcoin_status().await;
let body = serde_json::to_vec(&status).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.header("Cache-Control", "no-store")
.body(hyper::Body::from(body))
.unwrap_or_else(|_| Response::new(hyper::Body::from("{}"))))
}
pub(super) async fn handle_lnd_connect_info(
rpc: std::sync::Arc<super::super::rpc::RpcHandler>,
cors_origin: &str,
) -> Result<Response<hyper::Body>> {
// The LND wallet UI is served on its own APP_PORTS origin and fetches
// this cross-origin, so it needs the CORS headers echoed back.
let cors = |builder: hyper::http::response::Builder| {
builder
.header("Access-Control-Allow-Origin", cors_origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
};
match rpc.handle_lnd_connect_info().await {
Ok(val) => {
let body = serde_json::to_vec(&val).unwrap_or_default();
Ok(cors(
Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json"),
)
.body(hyper::Body::from(body))
.unwrap_or_else(|_| Response::new(hyper::Body::from("{}"))))
}
Err(e) => Ok(cors(
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.header("Content-Type", "application/json"),
)
.body(hyper::Body::from(
serde_json::json!({"error": e.to_string()}).to_string(),
))
.unwrap()),
}
}
pub(super) async fn handle_lnd_proxy(
rpc: Arc<RpcHandler>,
path: &str,
cors_origin: &str,
) -> Result<Response<hyper::Body>> {
let suffix = path.strip_prefix("/proxy/lnd").unwrap_or("/");
let url = format!("{LND_REST_BASE_URL}{suffix}");
// LND REST serves a self-signed cert and requires the admin macaroon.
// A bare reqwest::get() uses the default client, which rejects the
// self-signed cert (TLS verify error -> 502 "failing to fetch") and
// sends no macaroon. Use the shared authenticated client instead — the
// same one lnd.getinfo and the wallet RPCs use.
let request = match rpc.lnd_client().await {
Ok((client, macaroon_hex)) => client
.get(&url)
.header("Grpc-Metadata-macaroon", &macaroon_hex)
.send()
.await
.map_err(anyhow::Error::from),
Err(e) => Err(e),
};
match request {
Ok(resp) => {
let status = resp.status().as_u16();
let headers = resp.headers().clone();
let body = resp.bytes().await.unwrap_or_default();
let mut builder = Response::builder().status(status);
if let Some(ct) = headers.get("content-type") {
if let Ok(s) = ct.to_str() {
builder = builder.header("Content-Type", s);
}
}
builder
.header("Access-Control-Allow-Origin", cors_origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
.body(hyper::Body::from(body))
.map_err(|e| anyhow::anyhow!("response build: {}", e))
}
Err(e) => {
let body = serde_json::json!({ "error": e.to_string() });
let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
Ok(Response::builder()
.status(StatusCode::BAD_GATEWAY)
.header("Content-Type", "application/json")
.header("Access-Control-Allow-Origin", cors_origin)
.header("Access-Control-Allow-Credentials", "true")
.header("Vary", "Origin")
.body(hyper::Body::from(body_bytes))
.unwrap())
}
}
}
/// Range-streaming proxy for a peer's content file (B3). The browser's
/// `<video>`/`<audio>` element makes Range requests; we forward the Range
/// header to the peer's `/content/<id>` (which already returns 206 Partial
/// Content) and pass the bytes + Content-Range/Content-Type straight back.
/// This replaces the old path of downloading the whole file as base64 into
/// a non-seekable Blob URL, which broke playback/seeking for video and
/// large audio. Same-origin + session-authenticated (checked by caller).
/// Path: `/api/peer-content/<onion>/<content_id>`.
pub(super) async fn handle_peer_content_stream(
&self,
path: &str,
headers: &hyper::HeaderMap,
) -> Result<Response<hyper::Body>> {
let bad = |msg: &str| {
Ok(build_response(
StatusCode::BAD_REQUEST,
"application/json",
hyper::Body::from(serde_json::json!({ "error": msg }).to_string()),
))
};
let rest = path.strip_prefix("/api/peer-content/").unwrap_or("");
let (onion, content_id) = match rest.split_once('/') {
Some((o, c)) if !o.is_empty() && !c.is_empty() => (o, c),
_ => return bad("expected /api/peer-content/<onion>/<content_id>"),
};
// Validate to prevent SSRF / path traversal.
let onion_norm = onion.trim_end_matches(".onion");
let onion_ok = onion_norm.len() == 56
&& onion_norm
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit());
let id_ok = !content_id.contains("..")
&& content_id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'));
if !onion_ok || !id_ok {
return bad("invalid onion or content id");
}
let fips_npub = crate::federation::fips_npub_for_onion(&self.config.data_dir, onion).await;
let peer_path = format!("/content/{}", content_id);
// Generous overall timeout: this endpoint serves both seek/Range
// playback (small, finishes fast) and full-file downloads of large
// media (#38). 60s was too tight for a multi-hundred-MB transfer over
// Tor and aborted the download mid-stream.
let mut req = crate::fips::dial::PeerRequest::new(fips_npub.as_deref(), onion, &peer_path)
.service(crate::settings::transport::PeerService::PeerFiles)
.timeout(std::time::Duration::from_secs(900));
if let Some(r) = headers.get("range").and_then(|v| v.to_str().ok()) {
req = req.header("Range", r.to_string());
}
match req.send_get().await {
Ok((resp, _transport)) => {
let status = resp.status().as_u16();
let rh = resp.headers().clone();
let mut builder = Response::builder()
.status(status)
.header("Accept-Ranges", "bytes");
for h in ["content-type", "content-range", "content-length"] {
if let Some(v) = rh.get(h).and_then(|v| v.to_str().ok()) {
builder = builder.header(h, v);
}
}
// Stream the peer's body straight through instead of buffering
// the whole file into memory (#38). For a 178MB download the old
// `resp.bytes().await` allocated the entire file on the node
// before sending a byte; `wrap_stream` forwards chunks as they
// arrive, with constant memory.
Ok(builder
.body(hyper::Body::wrap_stream(resp.bytes_stream()))
.unwrap_or_else(|_| Response::new(hyper::Body::empty())))
}
Err(e) => Ok(build_response(
StatusCode::BAD_GATEWAY,
"application/json",
hyper::Body::from(serde_json::json!({ "error": e.to_string() }).to_string()),
)),
}
}
}
core/archipelago/src/api/handler/remote_input.rs
+362
View File
@@ -0,0 +1,362 @@
use anyhow::{Context, Result};
use futures_util::{SinkExt, StreamExt};
use hyper::{Request, Response};
use hyper_ws_listener::WsStream;
use serde::Deserialize;
use std::time::Instant;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info, warn};
use super::ApiHandler;
/// Allowed xdotool key names. Only these pass validation.
const ALLOWED_KEYS: &[&str] = &[
// Letters
"a",
"b",
"c",
"d",
"e",
"f",
"g",
"h",
"i",
"j",
"k",
"l",
"m",
"n",
"o",
"p",
"q",
"r",
"s",
"t",
"u",
"v",
"w",
"x",
"y",
"z",
"A",
"B",
"C",
"D",
"E",
"F",
"G",
"H",
"I",
"J",
"K",
"L",
"M",
"N",
"O",
"P",
"Q",
"R",
"S",
"T",
"U",
"V",
"W",
"X",
"Y",
"Z",
// Numbers
"0",
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
// Navigation
"Up",
"Down",
"Left",
"Right",
"Return",
"Escape",
"Tab",
"BackSpace",
"Delete",
"Home",
"End",
"Prior",
"Next", // Prior=PageUp, Next=PageDown
// Modifiers (for combos like shift+a)
"space",
"minus",
"equal",
"bracketleft",
"bracketright",
"backslash",
"semicolon",
"apostrophe",
"grave",
"comma",
"period",
"slash",
// Function keys
"F1",
"F2",
"F3",
"F4",
"F5",
"F6",
"F7",
"F8",
"F9",
"F10",
"F11",
"F12",
// Symbols — xdotool names
"exclam",
"at",
"numbersign",
"dollar",
"percent",
"asciicircum",
"ampersand",
"asterisk",
"parenleft",
"parenright",
"underscore",
"plus",
"braceleft",
"braceright",
"bar",
"colon",
"quotedbl",
"less",
"greater",
"question",
"asciitilde",
];
/// Validate a key name against the whitelist.
/// Also allows "shift+X" combos where X is in the whitelist.
fn validate_key(key: &str) -> bool {
if ALLOWED_KEYS.contains(&key) {
return true;
}
// Allow modifier combos: "shift+a", "ctrl+c", etc.
if let Some((modifier, base)) = key.split_once('+') {
let valid_modifiers = ["shift", "ctrl", "alt", "super"];
return valid_modifiers.contains(&modifier) && ALLOWED_KEYS.contains(&base);
}
false
}
#[derive(Deserialize)]
#[serde(tag = "t")]
enum InputCommand {
#[serde(rename = "k")]
Key {
k: String,
/// Optional player ID (1 or 2) for multi-player arcade games.
/// When absent, input is broadcast without player tagging.
#[serde(default)]
#[allow(dead_code)]
p: Option<u8>,
},
#[serde(rename = "m")]
MouseMove { x: i32, y: i32 },
#[serde(rename = "c")]
Click { b: u8 },
#[serde(rename = "s")]
Scroll { y: i32 },
#[serde(rename = "p")]
Ping,
}
/// Validate and acknowledge input — relay-only, no xdotool.
/// All input is forwarded to browser clients via the broadcast channel;
/// the browser's remote-relay.ts dispatches DOM events from there.
async fn handle_input(msg: &str) -> Result<Option<String>> {
let cmd: InputCommand = serde_json::from_str(msg).context("invalid input command")?;
match cmd {
InputCommand::Key { ref k, .. } => {
if !validate_key(k) {
warn!("rejected key: {}", k);
return Ok(Some(r#"{"t":"e","m":"invalid key"}"#.to_string()));
}
}
InputCommand::MouseMove { x, y } => {
let _x = x.clamp(-50, 50);
let _y = y.clamp(-50, 50);
}
InputCommand::Click { b } => {
let _b = b.clamp(1, 3);
}
InputCommand::Scroll { y } => {
let _y = y.clamp(-10, 10);
}
InputCommand::Ping => {
return Ok(Some(r#"{"t":"p"}"#.to_string()));
}
}
Ok(None)
}
impl ApiHandler {
pub(super) async fn handle_remote_input(
req: Request<hyper::Body>,
relay_tx: broadcast::Sender<String>,
mut external_open_rx: broadcast::Receiver<String>,
) -> Result<Response<hyper::Body>> {
// Extract optional player ID from query string: /ws/remote-input?p=1
let player_id: Option<u8> = req
.uri()
.query()
.and_then(|q| q.split('&').find(|s| s.starts_with("p=")))
.and_then(|s| s.get(2..))
.and_then(|v| v.parse().ok())
.filter(|&p: &u8| p == 1 || p == 2);
let (response, ws_fut_opt) = hyper_ws_listener::create_ws(req)
.map_err(|e| anyhow::anyhow!("WebSocket upgrade failed: {}", e))?;
if let Some(ws_fut) = ws_fut_opt {
tokio::spawn(async move {
let ws_stream: WsStream = match ws_fut.await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
debug!("Remote input WS handshake failed: {}", e);
return;
}
Err(e) => {
debug!("Remote input WS task join failed: {}", e);
return;
}
};
info!("Remote input connected");
let (mut tx, mut rx) = ws_stream.split();
// Send ready message
let _ = tx.send(Message::Text(r#"{"t":"ok"}"#.to_string())).await;
let ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
tokio::pin!(ping_interval);
let mut last_activity = Instant::now();
let mut msg_count: u64 = 0;
let mut rate_window_start = Instant::now();
let mut rate_count: u32 = 0;
const MAX_RATE: u32 = 120; // messages per second
const INACTIVITY_TIMEOUT: u64 = 300;
loop {
tokio::select! {
_ = ping_interval.tick() => {
if last_activity.elapsed().as_secs() >= INACTIVITY_TIMEOUT {
info!("Remote input inactive, closing");
let _ = tx.send(Message::Close(None)).await;
break;
}
if tx.send(Message::Ping(vec![])).await.is_err() {
break;
}
}
// Forward kiosk "open this URL externally" requests down to
// the companion so the link opens in the phone's browser.
ext = external_open_rx.recv() => {
match ext {
Ok(text) => {
if tx.send(Message::Text(text)).await.is_err() {
break;
}
}
Err(broadcast::error::RecvError::Lagged(_)) => {}
Err(broadcast::error::RecvError::Closed) => {}
}
}
msg = rx.next() => {
match msg {
Some(Ok(Message::Text(text))) => {
last_activity = Instant::now();
msg_count += 1;
// Rate limiting
if rate_window_start.elapsed().as_millis() >= 1000 {
rate_window_start = Instant::now();
rate_count = 0;
}
rate_count += 1;
if rate_count > MAX_RATE {
continue; // silently drop
}
// Relay to browser clients. If this connection has a
// player ID from query string and the message is a key
// event without a player field, inject it so the browser
// can route input to the correct player.
let relay_text = if let Some(pid) = player_id {
if text.contains(r#""t":"k""#) && !text.contains(r#""p":"#) {
// Insert "p":N before the closing brace
if let Some(pos) = text.rfind('}') {
let mut tagged = text[..pos].to_string();
tagged.push_str(&format!(r#","p":{}"#, pid));
tagged.push('}');
tagged
} else {
text.clone()
}
} else {
text.clone()
}
} else {
text.clone()
};
let _ = relay_tx.send(relay_text);
match handle_input(&text).await {
Ok(Some(reply)) => {
let _ = tx.send(Message::Text(reply)).await;
}
Ok(None) => {}
Err(e) => {
debug!("Input error: {}", e);
let err = format!(r#"{{"t":"e","m":"{}"}}"#,
e.to_string().replace('"', "'"));
let _ = tx.send(Message::Text(err)).await;
}
}
}
Some(Ok(Message::Pong(_))) => {
last_activity = Instant::now();
}
Some(Ok(Message::Ping(data))) => {
last_activity = Instant::now();
let _ = tx.send(Message::Pong(data)).await;
}
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(_)) => { last_activity = Instant::now(); }
Some(Err(e)) => {
debug!("Remote input stream error: {}", e);
break;
}
}
}
}
}
info!(
"Remote input disconnected ({} messages processed)",
msg_count
);
});
}
Ok(response)
}
}
core/archipelago/src/api/handler/remote_relay.rs
+126
View File
@@ -0,0 +1,126 @@
use anyhow::Result;
use futures_util::{SinkExt, StreamExt};
use hyper::{Request, Response};
use hyper_ws_listener::WsStream;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info};
use super::ApiHandler;
impl ApiHandler {
/// WebSocket endpoint for browser clients to receive relayed companion input.
/// The browser's remote-relay.ts dispatches these as DOM keyboard/mouse events.
///
/// The kiosk also uses this socket in the *reverse* direction: when an "open
/// in external browser" app is launched, the kiosk can't usefully open it
/// itself, so it sends `{"t":"o","url":"https://…"}` here. We validate the
/// URL and publish it on `external_open_tx`, which the companion (phone)
/// socket forwards so the link opens in the phone's default browser.
pub(super) async fn handle_remote_relay(
req: Request<hyper::Body>,
mut relay_rx: broadcast::Receiver<String>,
external_open_tx: broadcast::Sender<String>,
) -> Result<Response<hyper::Body>> {
let (response, ws_fut_opt) = hyper_ws_listener::create_ws(req)
.map_err(|e| anyhow::anyhow!("WebSocket upgrade failed: {}", e))?;
if let Some(ws_fut) = ws_fut_opt {
tokio::spawn(async move {
let ws_stream: WsStream = match ws_fut.await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
debug!("Remote relay WS handshake failed: {}", e);
return;
}
Err(e) => {
debug!("Remote relay WS task join failed: {}", e);
return;
}
};
info!("Remote relay client connected");
let (mut tx, mut rx) = ws_stream.split();
// Send ready message
let _ = tx.send(Message::Text(r#"{"t":"ok"}"#.to_string())).await;
let ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
tokio::pin!(ping_interval);
loop {
tokio::select! {
_ = ping_interval.tick() => {
if tx.send(Message::Ping(vec![])).await.is_err() {
break;
}
}
// Forward relayed input from companion app
msg = relay_rx.recv() => {
match msg {
Ok(text) => {
if tx.send(Message::Text(text)).await.is_err() {
break;
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
debug!("Remote relay lagged, dropped {} messages", n);
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
// Handle client-side messages (pong, close, open-url requests)
client_msg = rx.next() => {
match client_msg {
Some(Ok(Message::Pong(_))) | Some(Ok(Message::Ping(_))) => {}
Some(Ok(Message::Text(text))) => {
// The only kiosk→server message we accept is an
// external-open request: {"t":"o","url":"https://…"}.
if let Some(url) = parse_open_url(&text) {
debug!("Relaying external-open to companion: {}", url);
let _ = external_open_tx.send(
format!(r#"{{"t":"o","url":{}}}"#, json_string(&url))
);
}
}
Some(Ok(Message::Close(_))) | None => break,
_ => {}
}
}
}
}
info!("Remote relay client disconnected");
});
}
Ok(response)
}
}
/// Parse a kiosk `{"t":"o","url":"…"}` external-open request, returning the URL
/// only if it's a well-formed http(s) URL. Anything else (other message tags,
/// non-http schemes like `javascript:`/`file:`, malformed JSON) is rejected so a
/// compromised kiosk page can't push arbitrary URIs to the phone.
fn parse_open_url(text: &str) -> Option<String> {
let v: serde_json::Value = serde_json::from_str(text).ok()?;
if v.get("t").and_then(|t| t.as_str()) != Some("o") {
return None;
}
let url = v.get("url").and_then(|u| u.as_str())?.trim();
if url.len() > 2048 {
return None;
}
let lower = url.to_ascii_lowercase();
if lower.starts_with("http://") || lower.starts_with("https://") {
Some(url.to_string())
} else {
None
}
}
/// Serialize a string as a JSON string literal (with surrounding quotes).
fn json_string(s: &str) -> String {
serde_json::Value::String(s.to_string()).to_string()
}
core/archipelago/src/api/handler/websocket.rs
+137
View File
@@ -0,0 +1,137 @@
use crate::monitoring::MetricsStore;
use crate::state::StateManager;
use anyhow::Result;
use futures_util::{SinkExt, StreamExt};
use hyper::{Request, Response};
use hyper_ws_listener::WsStream;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info};
use super::ApiHandler;
impl ApiHandler {
pub(super) async fn handle_websocket(
req: Request<hyper::Body>,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
) -> Result<Response<hyper::Body>> {
let (response, ws_fut_opt) = hyper_ws_listener::create_ws(req)
.map_err(|e| anyhow::anyhow!("WebSocket upgrade failed: {}", e))?;
if let Some(ws_fut) = ws_fut_opt {
tokio::spawn(async move {
let ws_stream: WsStream = match ws_fut.await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
debug!("WebSocket handshake failed (hyper): {}", e);
return;
}
Err(e) => {
debug!("WebSocket task join failed: {}", e);
return;
}
};
metrics_store.increment_ws();
info!("WebSocket /ws/db connected");
let (mut tx, mut rx) = ws_stream.split();
// Subscribe BEFORE taking the initial snapshot. Messages are full
// data dumps keyed by a monotonic revision, so a broadcast that
// races the snapshot is at worst a harmless duplicate/newer dump
// delivered right after — but subscribing after the snapshot send
// (the old order) let any update in that window vanish forever,
// since a tokio broadcast channel never delivers sends that
// predate subscribe(). That silently stuck clients (e.g. a fresh
// install's post-boot container scan) on a stale initial snapshot
// until a full page reload opened a new connection past the race.
let mut state_rx = state_manager.subscribe();
let initial_msg = state_manager.get_initial_message().await;
if let Ok(json_msg) = serde_json::to_string(&initial_msg) {
if let Err(e) = tx.send(Message::Text(json_msg)).await {
debug!("Failed to send initial data: {}", e);
return;
}
debug!("Sent initial data dump at revision {}", initial_msg.rev);
}
let ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
tokio::pin!(ping_interval);
let mut last_client_activity = Instant::now();
const INACTIVITY_TIMEOUT_SECS: u64 = 300; // 5 minutes
loop {
tokio::select! {
_ = ping_interval.tick() => {
// Check inactivity timeout
if last_client_activity.elapsed().as_secs() >= INACTIVITY_TIMEOUT_SECS {
info!("WebSocket client inactive for {}s, closing", INACTIVITY_TIMEOUT_SECS);
let _ = tx.send(Message::Close(None)).await;
break;
}
if tx.send(Message::Ping(vec![])).await.is_err() {
debug!("Failed to send ping, connection likely closed");
break;
}
}
update = state_rx.recv() => {
match update {
Ok(msg) => {
if let Ok(json_msg) = serde_json::to_string(&msg) {
if let Err(e) = tx.send(Message::Text(json_msg)).await {
debug!("Failed to send state update: {}", e);
break;
}
debug!("Sent state update at revision {}", msg.rev);
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
debug!("Client lagged behind, skipped {} messages", skipped);
}
Err(broadcast::error::RecvError::Closed) => {
debug!("Broadcast channel closed");
break;
}
}
}
msg = rx.next() => {
match msg {
Some(Ok(Message::Close(_))) => break,
Some(Ok(Message::Pong(_))) => {
last_client_activity = Instant::now();
debug!("Received pong");
}
Some(Ok(Message::Ping(data))) => {
last_client_activity = Instant::now();
let _ = tx.send(Message::Pong(data)).await;
}
Some(Ok(Message::Text(text))) => {
last_client_activity = Instant::now();
// Handle JSON ping from frontend
if text.contains("\"type\":\"ping\"") || text.contains("\"type\": \"ping\"") {
let _ = tx.send(Message::Text(r#"{"type":"pong"}"#.to_string())).await;
}
}
Some(Ok(_)) => {
last_client_activity = Instant::now();
}
Some(Err(e)) => {
debug!("WebSocket stream error: {}", e);
break;
}
None => break,
}
}
}
}
metrics_store.decrement_ws();
info!("WebSocket /ws/db disconnected");
});
}
Ok(response)
}
}