feat: phase 7 — proxy daemon with HTTP login and WebSocket bridge
ericrfb-proxy axum server with three endpoints: POST /api/login: - Proxies credentials to OmniView auth.asp - Extracts session cookie, fetches title_app.asp - Returns JSON with applet_id, port, protocol_version, board_name GET /api/ws?applet_id=...&port=...: - WebSocket upgrade, connects to OmniView via e-RIC RFB - Bidirectional pump: OmniView frames → RGBA blits over WS, browser input events → key/mouse/hotkey to OmniView - Binary protocol: TAG_BLIT(0x01), TAG_RESIZE(0x03) server→client; TAG_KEY_PRESS(0x10), TAG_KEY_RELEASE(0x11), TAG_POINTER(0x12), TAG_CTRL_ALT_DEL(0x13) client→server Static file fallback via tower-http ServeDir. Config via config.toml or BLEKIN_HOST env var. Tested against real OmniView: - Login endpoint returns valid APPLET_ID - WebSocket upgrade succeeds (HTTP 101) - Session connects and pumps frames Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -5,8 +5,13 @@ edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
ericrfb = { path = "../ericrfb" }
|
||||
futures-util = "0.3"
|
||||
tokio.workspace = true
|
||||
axum.workspace = true
|
||||
reqwest.workspace = true
|
||||
serde.workspace = true
|
||||
toml.workspace = true
|
||||
tower-http.workspace = true
|
||||
tracing.workspace = true
|
||||
tracing-subscriber.workspace = true
|
||||
anyhow.workspace = true
|
||||
|
||||
64
crates/ericrfb-proxy/src/config.rs
Normal file
64
crates/ericrfb-proxy/src/config.rs
Normal file
@@ -0,0 +1,64 @@
|
||||
use serde::Deserialize;
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct ProxyConfig {
|
||||
/// Address to bind the proxy HTTP server to.
|
||||
#[serde(default = "default_bind")]
|
||||
pub bind: String,
|
||||
|
||||
/// Directory to serve static frontend files from.
|
||||
#[serde(default = "default_static_dir")]
|
||||
pub static_dir: String,
|
||||
|
||||
pub omniview: OmniviewConfig,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct OmniviewConfig {
|
||||
pub host: String,
|
||||
|
||||
/// HTTP port for login/applet page (usually 80).
|
||||
#[serde(default = "default_http_port")]
|
||||
pub http_port: u16,
|
||||
|
||||
/// TCP port for e-RIC RFB protocol (usually 443).
|
||||
#[serde(default = "default_rfb_port")]
|
||||
pub rfb_port: u16,
|
||||
}
|
||||
|
||||
fn default_bind() -> String {
|
||||
"0.0.0.0:3000".into()
|
||||
}
|
||||
|
||||
fn default_static_dir() -> String {
|
||||
"dist".into()
|
||||
}
|
||||
|
||||
fn default_http_port() -> u16 {
|
||||
80
|
||||
}
|
||||
|
||||
fn default_rfb_port() -> u16 {
|
||||
443
|
||||
}
|
||||
|
||||
pub fn load() -> anyhow::Result<ProxyConfig> {
|
||||
let path = std::env::var("BLEKIN_CONFIG").unwrap_or_else(|_| "config.toml".into());
|
||||
match std::fs::read_to_string(&path) {
|
||||
Ok(contents) => Ok(toml::from_str(&contents)?),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
|
||||
tracing::warn!("no config file at {path}, using defaults + BLEKIN_HOST env");
|
||||
let host = std::env::var("BLEKIN_HOST").unwrap_or_else(|_| "10.3.0.130".into());
|
||||
Ok(ProxyConfig {
|
||||
bind: default_bind(),
|
||||
static_dir: default_static_dir(),
|
||||
omniview: OmniviewConfig {
|
||||
host,
|
||||
http_port: default_http_port(),
|
||||
rfb_port: default_rfb_port(),
|
||||
},
|
||||
})
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
}
|
||||
116
crates/ericrfb-proxy/src/login.rs
Normal file
116
crates/ericrfb-proxy/src/login.rs
Normal file
@@ -0,0 +1,116 @@
|
||||
use axum::Json;
|
||||
use axum::extract::State;
|
||||
use axum::http::StatusCode;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::AppState;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct LoginRequest {
|
||||
pub username: String,
|
||||
pub password: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct LoginResponse {
|
||||
pub applet_id: String,
|
||||
pub port: u16,
|
||||
pub protocol_version: String,
|
||||
pub board_name: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct ErrorResponse {
|
||||
pub error: String,
|
||||
}
|
||||
|
||||
pub async fn handle_login(
|
||||
State(state): State<AppState>,
|
||||
Json(req): Json<LoginRequest>,
|
||||
) -> Result<Json<LoginResponse>, (StatusCode, Json<ErrorResponse>)> {
|
||||
let cfg = &state.config.omniview;
|
||||
let base = format!("http://{}:{}", cfg.host, cfg.http_port);
|
||||
|
||||
// POST credentials to auth.asp
|
||||
let auth_resp = state
|
||||
.http_client
|
||||
.post(format!("{base}/auth.asp"))
|
||||
.form(&[
|
||||
("login", req.username.as_str()),
|
||||
("password", req.password.as_str()),
|
||||
("action_login.x", "0"),
|
||||
("action_login.y", "0"),
|
||||
])
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| auth_err(format!("connect failed: {e}")))?;
|
||||
|
||||
// Extract session cookie
|
||||
let cookie = auth_resp
|
||||
.headers()
|
||||
.get("set-cookie")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|s| s.split(';').next())
|
||||
.ok_or_else(|| auth_err("no session cookie in response"))?
|
||||
.to_string();
|
||||
|
||||
// Check redirect — should go to home.asp on success
|
||||
let location = auth_resp
|
||||
.headers()
|
||||
.get("location")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
if !location.contains("home.asp") {
|
||||
return Err(auth_err("authentication failed"));
|
||||
}
|
||||
|
||||
// Fetch the applet page to extract params
|
||||
let applet_resp = state
|
||||
.http_client
|
||||
.get(format!("{base}/title_app.asp"))
|
||||
.header("cookie", &cookie)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| auth_err(format!("failed to fetch applet page: {e}")))?;
|
||||
|
||||
let html = applet_resp
|
||||
.text()
|
||||
.await
|
||||
.map_err(|e| auth_err(format!("failed to read applet page: {e}")))?;
|
||||
|
||||
let applet_id = extract_param(&html, "APPLET_ID")
|
||||
.ok_or_else(|| auth_err("APPLET_ID not found in applet page"))?;
|
||||
let port = extract_param(&html, "PORT")
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(cfg.rfb_port);
|
||||
let protocol_version =
|
||||
extract_param(&html, "PROTOCOL_VERSION").unwrap_or_else(|| "01.11".into());
|
||||
let board_name =
|
||||
extract_param(&html, "BOARD_NAME").unwrap_or_else(|| "Remote IP Manager".into());
|
||||
|
||||
tracing::info!(
|
||||
"login successful: board={board_name}, applet_id={}...",
|
||||
&applet_id[..applet_id.len().min(16)]
|
||||
);
|
||||
|
||||
Ok(Json(LoginResponse {
|
||||
applet_id,
|
||||
port,
|
||||
protocol_version,
|
||||
board_name,
|
||||
}))
|
||||
}
|
||||
|
||||
fn extract_param(html: &str, name: &str) -> Option<String> {
|
||||
let needle = format!("{name}\" value=\"");
|
||||
let start = html.find(&needle)? + needle.len();
|
||||
let end = html[start..].find('"')? + start;
|
||||
Some(html[start..end].to_string())
|
||||
}
|
||||
|
||||
fn auth_err(msg: impl Into<String>) -> (StatusCode, Json<ErrorResponse>) {
|
||||
(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(ErrorResponse { error: msg.into() }),
|
||||
)
|
||||
}
|
||||
@@ -1,3 +1,52 @@
|
||||
fn main() {
|
||||
println!("Hello, world!");
|
||||
mod config;
|
||||
mod login;
|
||||
mod ws;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::Router;
|
||||
use axum::routing::{get, post};
|
||||
use tokio::net::TcpListener;
|
||||
use tower_http::services::ServeDir;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct AppState {
|
||||
pub config: Arc<config::ProxyConfig>,
|
||||
pub http_client: reqwest::Client,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(EnvFilter::from_default_env())
|
||||
.init();
|
||||
|
||||
let cfg = config::load()?;
|
||||
tracing::info!(
|
||||
"blekin proxy starting — OmniView at {}:{}, binding to {}",
|
||||
cfg.omniview.host,
|
||||
cfg.omniview.http_port,
|
||||
cfg.bind
|
||||
);
|
||||
|
||||
let state = AppState {
|
||||
config: Arc::new(cfg.clone()),
|
||||
http_client: reqwest::Client::builder()
|
||||
.danger_accept_invalid_certs(true)
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.build()?,
|
||||
};
|
||||
|
||||
let app = Router::new()
|
||||
.route("/api/login", post(login::handle_login))
|
||||
.route("/api/ws", get(ws::handle_ws))
|
||||
.fallback_service(ServeDir::new(&cfg.static_dir))
|
||||
.with_state(state);
|
||||
|
||||
let listener = TcpListener::bind(&cfg.bind).await?;
|
||||
tracing::info!("listening on {}", cfg.bind);
|
||||
axum::serve(listener, app).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
256
crates/ericrfb-proxy/src/ws.rs
Normal file
256
crates/ericrfb-proxy/src/ws.rs
Normal file
@@ -0,0 +1,256 @@
|
||||
use axum::extract::ws::{Message, WebSocket};
|
||||
use axum::extract::{Query, State, WebSocketUpgrade};
|
||||
use axum::response::IntoResponse;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde::Deserialize;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use ericrfb::framebuffer::Framebuffer;
|
||||
use ericrfb::handshake::Config;
|
||||
use ericrfb::input;
|
||||
use ericrfb::msg;
|
||||
use ericrfb::proto::RGB332_LUT;
|
||||
use ericrfb::session::{ActiveSession, Event};
|
||||
|
||||
use crate::AppState;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// WS binary protocol tags
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// Proxy → Browser
|
||||
const TAG_BLIT: u8 = 0x01;
|
||||
const TAG_RESIZE: u8 = 0x03;
|
||||
|
||||
// Browser → Proxy
|
||||
const TAG_KEY_PRESS: u8 = 0x10;
|
||||
const TAG_KEY_RELEASE: u8 = 0x11;
|
||||
const TAG_POINTER: u8 = 0x12;
|
||||
const TAG_CTRL_ALT_DEL: u8 = 0x13;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct WsQuery {
|
||||
pub applet_id: String,
|
||||
#[serde(default = "default_port")]
|
||||
pub port: u16,
|
||||
}
|
||||
|
||||
fn default_port() -> u16 {
|
||||
443
|
||||
}
|
||||
|
||||
pub async fn handle_ws(
|
||||
ws: WebSocketUpgrade,
|
||||
State(state): State<AppState>,
|
||||
Query(query): Query<WsQuery>,
|
||||
) -> impl IntoResponse {
|
||||
ws.on_upgrade(move |socket| run_session(socket, state, query))
|
||||
}
|
||||
|
||||
async fn run_session(socket: WebSocket, state: AppState, query: WsQuery) {
|
||||
let cfg = Config::new(&state.config.omniview.host, query.port, &query.applet_id);
|
||||
|
||||
tracing::info!(
|
||||
"WS session starting: {}:{}",
|
||||
state.config.omniview.host,
|
||||
query.port
|
||||
);
|
||||
|
||||
// Connect to OmniView in a blocking task (handshake is sync IO)
|
||||
let session = match tokio::task::spawn_blocking(move || {
|
||||
ActiveSession::connect(&cfg, &[7, 5, 1, 0, -250])
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(Ok(s)) => s,
|
||||
Ok(Err(e)) => {
|
||||
tracing::error!("OmniView connect failed: {e}");
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("spawn_blocking panicked: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
"Connected to OmniView: {}x{}",
|
||||
session.framebuffer.width,
|
||||
session.framebuffer.height
|
||||
);
|
||||
|
||||
let (ws_tx, ws_rx) = socket.split();
|
||||
let (blit_tx, blit_rx) = mpsc::channel::<Message>(64);
|
||||
|
||||
// Channel for input events from browser → OmniView writer
|
||||
let (input_tx, input_rx) = mpsc::channel::<InputEvent>(64);
|
||||
|
||||
// Task: forward blit messages to WebSocket
|
||||
let ws_send_task = tokio::spawn(forward_ws_send(ws_tx, blit_rx));
|
||||
|
||||
// Task: receive input from WebSocket
|
||||
let ws_recv_task = tokio::spawn(forward_ws_recv(ws_rx, input_tx));
|
||||
|
||||
// Task: OmniView session pump (blocking)
|
||||
let pump_task = tokio::task::spawn_blocking(move || run_pump(session, blit_tx, input_rx));
|
||||
|
||||
// Wait for any task to finish (on error or disconnect)
|
||||
tokio::select! {
|
||||
r = ws_send_task => { tracing::debug!("ws_send finished: {r:?}"); }
|
||||
r = ws_recv_task => { tracing::debug!("ws_recv finished: {r:?}"); }
|
||||
r = pump_task => { tracing::debug!("pump finished: {r:?}"); }
|
||||
}
|
||||
|
||||
tracing::info!("WS session ended");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// OmniView pump (runs on blocking thread)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
enum InputEvent {
|
||||
KeyPress(u8),
|
||||
KeyRelease(u8),
|
||||
Pointer { x: u16, y: u16, mask: u8 },
|
||||
CtrlAltDel,
|
||||
}
|
||||
|
||||
fn run_pump(
|
||||
mut session: ActiveSession,
|
||||
blit_tx: mpsc::Sender<Message>,
|
||||
mut input_rx: mpsc::Receiver<InputEvent>,
|
||||
) {
|
||||
// Send initial resize message
|
||||
let w = session.framebuffer.width;
|
||||
let h = session.framebuffer.height;
|
||||
let _ = blit_tx.blocking_send(make_resize_msg(w, h));
|
||||
|
||||
loop {
|
||||
// Drain any pending input events
|
||||
while let Ok(evt) = input_rx.try_recv() {
|
||||
if let Err(e) = handle_input(&mut session, evt) {
|
||||
tracing::error!("input error: {e}");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Process one server message
|
||||
match session.process_one() {
|
||||
Ok(Some(Event::FramebufferDirty)) => {
|
||||
// Send full framebuffer as RGBA blit
|
||||
let msg = make_full_blit(&session.framebuffer);
|
||||
if blit_tx.blocking_send(msg).is_err() {
|
||||
return; // WS closed
|
||||
}
|
||||
// Request next update
|
||||
if let Err(e) = session.request_update() {
|
||||
tracing::error!("request_update error: {e}");
|
||||
return;
|
||||
}
|
||||
}
|
||||
Ok(Some(Event::Resize { width, height })) => {
|
||||
let _ = blit_tx.blocking_send(make_resize_msg(width, height));
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
tracing::error!("session error: {e}");
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_input(session: &mut ActiveSession, evt: InputEvent) -> Result<(), String> {
|
||||
match evt {
|
||||
InputEvent::KeyPress(sc) => {
|
||||
input::write_key_press(&mut session.writer, sc).map_err(|e| e.to_string())
|
||||
}
|
||||
InputEvent::KeyRelease(sc) => {
|
||||
input::write_key_release(&mut session.writer, sc).map_err(|e| e.to_string())
|
||||
}
|
||||
InputEvent::Pointer { x, y, mask } => {
|
||||
msg::write_pointer_event(&mut session.writer, x, y, mask).map_err(|e| e.to_string())
|
||||
}
|
||||
InputEvent::CtrlAltDel => {
|
||||
input::write_ctrl_alt_del(&mut session.writer).map_err(|e| e.to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Binary message builders
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn make_full_blit(fb: &Framebuffer) -> Message {
|
||||
let w = fb.width;
|
||||
let h = fb.height;
|
||||
// Header: tag(1) + x(2) + y(2) + w(2) + h(2) = 9 bytes
|
||||
let mut buf = Vec::with_capacity(9 + (w as usize * h as usize * 4));
|
||||
buf.push(TAG_BLIT);
|
||||
buf.extend_from_slice(&0u16.to_be_bytes()); // x
|
||||
buf.extend_from_slice(&0u16.to_be_bytes()); // y
|
||||
buf.extend_from_slice(&w.to_be_bytes());
|
||||
buf.extend_from_slice(&h.to_be_bytes());
|
||||
// RGBA pixels
|
||||
for &px in &fb.pixels {
|
||||
buf.extend_from_slice(&RGB332_LUT[px as usize]);
|
||||
}
|
||||
Message::Binary(buf)
|
||||
}
|
||||
|
||||
fn make_resize_msg(w: u16, h: u16) -> Message {
|
||||
let mut buf = Vec::with_capacity(5);
|
||||
buf.push(TAG_RESIZE);
|
||||
buf.extend_from_slice(&w.to_be_bytes());
|
||||
buf.extend_from_slice(&h.to_be_bytes());
|
||||
Message::Binary(buf)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// WebSocket forwarding tasks
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async fn forward_ws_send(
|
||||
mut tx: futures_util::stream::SplitSink<WebSocket, Message>,
|
||||
mut rx: mpsc::Receiver<Message>,
|
||||
) {
|
||||
while let Some(msg) = rx.recv().await {
|
||||
if tx.send(msg).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn forward_ws_recv(
|
||||
mut rx: futures_util::stream::SplitStream<WebSocket>,
|
||||
tx: mpsc::Sender<InputEvent>,
|
||||
) {
|
||||
while let Some(Ok(msg)) = rx.next().await {
|
||||
match msg {
|
||||
Message::Binary(data) if !data.is_empty() => {
|
||||
if let Some(evt) = parse_input(&data)
|
||||
&& tx.send(evt).await.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
Message::Close(_) => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_input(data: &[u8]) -> Option<InputEvent> {
|
||||
match data[0] {
|
||||
TAG_KEY_PRESS if data.len() >= 2 => Some(InputEvent::KeyPress(data[1])),
|
||||
TAG_KEY_RELEASE if data.len() >= 2 => Some(InputEvent::KeyRelease(data[1])),
|
||||
TAG_POINTER if data.len() >= 6 => {
|
||||
let x = u16::from_be_bytes([data[1], data[2]]);
|
||||
let y = u16::from_be_bytes([data[3], data[4]]);
|
||||
let mask = data[5];
|
||||
Some(InputEvent::Pointer { x, y, mask })
|
||||
}
|
||||
TAG_CTRL_ALT_DEL => Some(InputEvent::CtrlAltDel),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user