aev_bridge/
lib.rs

1use anyhow::{anyhow, Context, Result};
2use axum::body::Body;
3use axum::extract::{Path, Query, State};
4use axum::http::{header, HeaderMap, HeaderValue, Request, StatusCode, Uri};
5use axum::middleware::{self, Next};
6use axum::response::sse::{Event, KeepAlive, Sse};
7use axum::response::{Html, IntoResponse, Response};
8use axum::routing::{get, post};
9use axum::{Json, Router};
10use clap::Parser;
11use futures_core::Stream;
12use reqwest::Method;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15use std::convert::Infallible;
16use std::ffi::c_char;
17use std::net::{SocketAddr, UdpSocket};
18use std::sync::Arc;
19use std::time::Duration;
20use tokio::net::TcpListener;
21use uuid::Uuid;
22
23pub const PLUGIN_ID: &str = "aev-bridge";
24pub const PLUGIN_NAME: &str = "aev-bridge";
25pub const CRATE_NAME: &str = "aev-bridge";
26pub const ENTRYPOINT: &str = "aev_plugin_init";
27pub const PROTOCOL: &str = "aev.plugin.v1";
28pub const DEFAULT_BIND: &str = "0.0.0.0:8780";
29pub const DEFAULT_UPSTREAM: &str = "http://127.0.0.1:8765";
30pub const PLUGIN_INIT_JSON: &str = r#"{"id":"aev-bridge","name":"aev-bridge","package":"aev-bridge","entrypoint":"aev_plugin_init","protocol":"aev.plugin.v1","capabilities":["tool","bridge","mobile","http","sse","bearer-auth"]}"#;
31const PLUGIN_INIT_JSON_NUL: &str = "{\"id\":\"aev-bridge\",\"name\":\"aev-bridge\",\"package\":\"aev-bridge\",\"entrypoint\":\"aev_plugin_init\",\"protocol\":\"aev.plugin.v1\",\"capabilities\":[\"tool\",\"bridge\",\"mobile\",\"http\",\"sse\",\"bearer-auth\"]}\0";
32
33#[no_mangle]
34pub extern "C" fn aev_plugin_init() -> *const c_char {
35    PLUGIN_INIT_JSON_NUL.as_ptr().cast()
36}
37
38#[derive(Parser, Debug, Clone)]
39#[command(name = "aev-bridge", about = "Start the Aev phone bridge for aev-web.")]
40pub struct Cli {
41    #[arg(long, env = "AEV_BRIDGE_BIND", default_value = DEFAULT_BIND)]
42    pub bind: SocketAddr,
43    #[arg(long, env = "AEV_BRIDGE_TOKEN")]
44    pub token: Option<String>,
45    #[arg(long, env = "AEV_BRIDGE_UPSTREAM", default_value = DEFAULT_UPSTREAM)]
46    pub upstream: String,
47    #[arg(long, env = "AEV_BRIDGE_UPSTREAM_TOKEN")]
48    pub upstream_token: Option<String>,
49    #[arg(long, env = "AEV_BRIDGE_PUBLIC_URL")]
50    pub public_url: Option<String>,
51    #[arg(long, env = "AEV_BRIDGE_POLL_MS", default_value_t = 300)]
52    pub poll_ms: u64,
53}
54
55#[derive(Debug, Clone)]
56pub struct ServerConfig {
57    pub bind: SocketAddr,
58    pub token: String,
59    pub upstream: String,
60    pub upstream_token: String,
61    pub public_url: Option<String>,
62    pub poll_ms: u64,
63}
64
65impl From<Cli> for ServerConfig {
66    fn from(cli: Cli) -> Self {
67        let upstream_token = cli
68            .upstream_token
69            .or_else(|| std::env::var("AEV_WEB_TOKEN").ok())
70            .unwrap_or_default();
71        Self {
72            bind: cli.bind,
73            token: normalize_token(cli.token),
74            upstream: normalize_upstream(&cli.upstream),
75            upstream_token: upstream_token.trim().to_string(),
76            public_url: cli
77                .public_url
78                .map(|url| url.trim().trim_end_matches('/').to_string()),
79            poll_ms: cli.poll_ms.clamp(100, 5_000),
80        }
81    }
82}
83
84#[derive(Debug, Clone, Serialize)]
85pub struct PluginManifest {
86    pub id: &'static str,
87    pub name: &'static str,
88    pub package: &'static str,
89    pub entrypoint: &'static str,
90    pub protocol: &'static str,
91    pub capabilities: Vec<&'static str>,
92    pub default_bind: &'static str,
93    pub default_upstream: &'static str,
94}
95
96pub fn plugin_manifest() -> PluginManifest {
97    PluginManifest {
98        id: PLUGIN_ID,
99        name: PLUGIN_NAME,
100        package: CRATE_NAME,
101        entrypoint: ENTRYPOINT,
102        protocol: PROTOCOL,
103        capabilities: vec!["tool", "bridge", "mobile", "http", "sse", "bearer-auth"],
104        default_bind: DEFAULT_BIND,
105        default_upstream: DEFAULT_UPSTREAM,
106    }
107}
108
109pub async fn serve(config: ServerConfig) -> Result<()> {
110    if config.upstream_token.trim().is_empty() {
111        return Err(anyhow!(
112            "missing upstream aev-web token; set AEV_WEB_TOKEN or --upstream-token"
113        ));
114    }
115
116    let bridge_url = phone_url(&config);
117    let state = AppState::new(config)?;
118    let app = router(state.clone());
119    let listener = TcpListener::bind(state.bind)
120        .await
121        .with_context(|| format!("failed to bind aev-bridge to {}", state.bind))?;
122    let local_addr = listener
123        .local_addr()
124        .context("failed to read aev-bridge listener address")?;
125
126    eprintln!("aev-bridge listening on http://{local_addr}");
127    eprintln!("phone URL: {bridge_url}/?token={}", state.token);
128    eprintln!("upstream: {}", state.upstream);
129
130    axum::serve(listener, app)
131        .with_graceful_shutdown(shutdown_signal())
132        .await
133        .context("aev-bridge server failed")
134}
135
136pub fn router(state: AppState) -> Router {
137    let auth_token = state.token.clone();
138    let api = Router::new()
139        .route("/manifest", get(manifest_handler))
140        .route("/upstream/health", get(upstream_health_handler))
141        .route(
142            "/sessions",
143            get(list_sessions_handler).post(create_session_handler),
144        )
145        .route(
146            "/sessions/{id}",
147            get(get_session_handler).delete(delete_session_handler),
148        )
149        .route("/sessions/{id}/events", get(output_events_handler))
150        .route("/sessions/{id}/input", post(write_input_handler))
151        .route_layer(middleware::from_fn_with_state(auth_token, require_token));
152
153    Router::new()
154        .route("/", get(index_handler))
155        .route("/health", get(health_handler))
156        .nest("/api", api)
157        .with_state(state)
158}
159
160#[derive(Clone)]
161pub struct AppState {
162    bind: SocketAddr,
163    token: Arc<str>,
164    upstream: Arc<str>,
165    upstream_token: Arc<str>,
166    poll: Duration,
167    client: reqwest::Client,
168}
169
170impl AppState {
171    pub fn new(config: ServerConfig) -> Result<Self> {
172        Ok(Self {
173            bind: config.bind,
174            token: Arc::<str>::from(config.token),
175            upstream: Arc::<str>::from(normalize_upstream(&config.upstream)),
176            upstream_token: Arc::<str>::from(config.upstream_token),
177            poll: Duration::from_millis(config.poll_ms.clamp(100, 5_000)),
178            client: reqwest::Client::builder()
179                .build()
180                .context("failed to build bridge HTTP client")?,
181        })
182    }
183
184    fn upstream_url(&self, path: &str) -> String {
185        format!("{}{}", self.upstream, path)
186    }
187
188    fn upstream_auth(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
189        request.bearer_auth(self.upstream_token.as_ref())
190    }
191}
192
193#[derive(Debug, Deserialize)]
194pub struct EventQuery {
195    pub since: Option<u64>,
196}
197
198#[derive(Debug, Clone, Deserialize, Serialize)]
199pub struct UpstreamOutput {
200    pub id: Uuid,
201    pub cursor: u64,
202    pub next: u64,
203    pub data: String,
204    pub dropped: bool,
205    pub alive: bool,
206}
207
208#[derive(Debug, Serialize)]
209struct HealthResponse {
210    id: &'static str,
211    status: &'static str,
212    protocol: &'static str,
213    upstream: String,
214}
215
216#[derive(Debug, Serialize)]
217struct ErrorResponse {
218    error: String,
219}
220
221#[derive(Debug)]
222struct ApiError {
223    status: StatusCode,
224    message: String,
225}
226
227impl ApiError {
228    fn bad_gateway(message: impl Into<String>) -> Self {
229        Self {
230            status: StatusCode::BAD_GATEWAY,
231            message: message.into(),
232        }
233    }
234
235    fn upstream(error: reqwest::Error) -> Self {
236        Self::bad_gateway(error.to_string())
237    }
238}
239
240impl IntoResponse for ApiError {
241    fn into_response(self) -> Response {
242        (
243            self.status,
244            Json(ErrorResponse {
245                error: self.message,
246            }),
247        )
248            .into_response()
249    }
250}
251
252async fn index_handler() -> Html<&'static str> {
253    Html(INDEX_HTML)
254}
255
256async fn health_handler(State(state): State<AppState>) -> Json<HealthResponse> {
257    Json(HealthResponse {
258        id: PLUGIN_ID,
259        status: "ok",
260        protocol: PROTOCOL,
261        upstream: state.upstream.to_string(),
262    })
263}
264
265async fn manifest_handler() -> Json<PluginManifest> {
266    Json(plugin_manifest())
267}
268
269async fn upstream_health_handler(State(state): State<AppState>) -> Result<Response, ApiError> {
270    proxy_to_upstream(&state, Method::GET, "/health", None).await
271}
272
273async fn list_sessions_handler(State(state): State<AppState>) -> Result<Response, ApiError> {
274    proxy_to_upstream(&state, Method::GET, "/api/sessions", None).await
275}
276
277async fn create_session_handler(
278    State(state): State<AppState>,
279    Json(body): Json<Value>,
280) -> Result<Response, ApiError> {
281    proxy_to_upstream(&state, Method::POST, "/api/sessions", Some(body)).await
282}
283
284async fn get_session_handler(
285    State(state): State<AppState>,
286    Path(id): Path<Uuid>,
287) -> Result<Response, ApiError> {
288    proxy_to_upstream(&state, Method::GET, &format!("/api/sessions/{id}"), None).await
289}
290
291async fn write_input_handler(
292    State(state): State<AppState>,
293    Path(id): Path<Uuid>,
294    Json(body): Json<Value>,
295) -> Result<Response, ApiError> {
296    proxy_to_upstream(
297        &state,
298        Method::POST,
299        &format!("/api/sessions/{id}/input"),
300        Some(body),
301    )
302    .await
303}
304
305async fn delete_session_handler(
306    State(state): State<AppState>,
307    Path(id): Path<Uuid>,
308) -> Result<Response, ApiError> {
309    proxy_to_upstream(&state, Method::DELETE, &format!("/api/sessions/{id}"), None).await
310}
311
312async fn output_events_handler(
313    State(state): State<AppState>,
314    Path(id): Path<Uuid>,
315    Query(query): Query<EventQuery>,
316) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, ApiError> {
317    let mut cursor = query.since.unwrap_or_default();
318    let stream = async_stream::stream! {
319        loop {
320            match read_upstream_output(&state, id, cursor).await {
321                Ok(output) => {
322                    cursor = output.next;
323                    if output.dropped || !output.data.is_empty() || !output.alive {
324                        match serde_json::to_string(&output) {
325                            Ok(json) => yield Ok(Event::default().event("output").data(json)),
326                            Err(error) => {
327                                let event = Event::default()
328                                    .event("error")
329                                    .data(error.to_string());
330                                yield Ok(event);
331                                break;
332                            }
333                        }
334                    }
335                    if !output.alive {
336                        break;
337                    }
338                }
339                Err(error) => {
340                    let event = Event::default().event("error").data(error.message);
341                    yield Ok(event);
342                    break;
343                }
344            }
345            tokio::time::sleep(state.poll).await;
346        }
347    };
348
349    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
350}
351
352async fn proxy_to_upstream(
353    state: &AppState,
354    method: Method,
355    path: &str,
356    body: Option<Value>,
357) -> Result<Response, ApiError> {
358    let mut request = state.upstream_auth(state.client.request(method, state.upstream_url(path)));
359    if let Some(body) = body {
360        request = request.json(&body);
361    }
362
363    let response = request.send().await.map_err(ApiError::upstream)?;
364    let status =
365        StatusCode::from_u16(response.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
366    let content_type = response
367        .headers()
368        .get(reqwest::header::CONTENT_TYPE)
369        .and_then(|value| value.to_str().ok())
370        .and_then(|value| HeaderValue::from_str(value).ok());
371    let text = response.text().await.map_err(ApiError::upstream)?;
372    let mut response = (status, text).into_response();
373    if let Some(content_type) = content_type {
374        response
375            .headers_mut()
376            .insert(header::CONTENT_TYPE, content_type);
377    }
378    Ok(response)
379}
380
381async fn read_upstream_output(
382    state: &AppState,
383    id: Uuid,
384    since: u64,
385) -> Result<UpstreamOutput, ApiError> {
386    let response = state
387        .upstream_auth(
388            state
389                .client
390                .get(state.upstream_url(&format!("/api/sessions/{id}/output?since={since}"))),
391        )
392        .send()
393        .await
394        .map_err(ApiError::upstream)?;
395    if !response.status().is_success() {
396        return Err(ApiError::bad_gateway(format!(
397            "upstream output read failed: {}",
398            response.status()
399        )));
400    }
401    response
402        .json::<UpstreamOutput>()
403        .await
404        .map_err(ApiError::upstream)
405}
406
407async fn require_token(State(token): State<Arc<str>>, req: Request<Body>, next: Next) -> Response {
408    if request_token_matches(req.headers(), req.uri(), token.as_ref()) {
409        return next.run(req).await;
410    }
411    (
412        StatusCode::UNAUTHORIZED,
413        Json(ErrorResponse {
414            error: "missing or invalid bridge token".to_string(),
415        }),
416    )
417        .into_response()
418}
419
420pub fn request_token_matches(headers: &HeaderMap, uri: &Uri, token: &str) -> bool {
421    bearer_token_matches(headers, token) || query_token_matches(uri, token)
422}
423
424pub fn bearer_token_matches(headers: &HeaderMap, token: &str) -> bool {
425    let Some(value) = headers
426        .get(header::AUTHORIZATION)
427        .and_then(|value| value.to_str().ok())
428    else {
429        return false;
430    };
431    let Some(candidate) = value.strip_prefix("Bearer ") else {
432        return false;
433    };
434    constant_time_eq(candidate.as_bytes(), token.as_bytes())
435}
436
437fn query_token_matches(uri: &Uri, token: &str) -> bool {
438    uri.query()
439        .and_then(|query| {
440            query.split('&').find_map(|part| {
441                let (key, value) = part.split_once('=')?;
442                (key == "token").then_some(value)
443            })
444        })
445        .map(|candidate| constant_time_eq(candidate.as_bytes(), token.as_bytes()))
446        .unwrap_or(false)
447}
448
449fn normalize_token(token: Option<String>) -> String {
450    token
451        .map(|token| token.trim().to_string())
452        .filter(|token| !token.is_empty())
453        .unwrap_or_else(|| Uuid::now_v7().to_string())
454}
455
456pub fn normalize_upstream(upstream: &str) -> String {
457    upstream.trim().trim_end_matches('/').to_string()
458}
459
460fn phone_url(config: &ServerConfig) -> String {
461    if let Some(public_url) = &config.public_url {
462        return public_url.clone();
463    }
464    if config.bind.ip().is_unspecified() {
465        if let Some(url) = lan_url(config.bind.port()) {
466            return url;
467        }
468    }
469    format!("http://{}", config.bind)
470}
471
472fn lan_url(port: u16) -> Option<String> {
473    let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
474    socket.connect("8.8.8.8:80").ok()?;
475    let local_addr = socket.local_addr().ok()?;
476    Some(format!("http://{}:{port}", local_addr.ip()))
477}
478
479fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
480    if left.len() != right.len() {
481        return false;
482    }
483    left.iter()
484        .zip(right)
485        .fold(0_u8, |diff, (left, right)| diff | (left ^ right))
486        == 0
487}
488
489async fn shutdown_signal() {
490    let _ = tokio::signal::ctrl_c().await;
491}
492
493const INDEX_HTML: &str = r#"<!doctype html>
494<html lang="en">
495<head>
496  <meta charset="utf-8">
497  <meta name="viewport" content="width=device-width, initial-scale=1">
498  <title>aev-bridge</title>
499  <style>
500    :root { color-scheme: dark; font-family: Inter, ui-sans-serif, system-ui, sans-serif; background: #07090f; color: #f5f7fb; }
501    body { margin: 0; min-height: 100vh; background: linear-gradient(135deg, #07111f, #140b24 55%, #05070b); }
502    main { min-height: 100vh; display: grid; grid-template-rows: auto auto 1fr auto; }
503    header { padding: 22px 18px 14px; display: flex; justify-content: space-between; gap: 12px; align-items: end; }
504    h1 { margin: 0; font-size: clamp(28px, 10vw, 58px); letter-spacing: -0.06em; }
505    .tag { color: #9da7bd; font-size: 13px; }
506    .bar { display: grid; grid-template-columns: 1fr auto; gap: 9px; padding: 0 14px 12px; }
507    input, textarea, button { border: 1px solid #2f3a50; border-radius: 14px; background: #0e1420; color: #f5f7fb; font: inherit; }
508    input, textarea { padding: 13px 14px; min-width: 0; }
509    button { padding: 13px 15px; background: #6d5dfc; border-color: #877cff; font-weight: 800; }
510    button.secondary { background: #182033; border-color: #2f3a50; }
511    pre { margin: 0 14px; border: 1px solid #253047; border-radius: 18px; background: #02040a; padding: 14px; overflow: auto; white-space: pre-wrap; word-break: break-word; font: 13px/1.45 ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; color: #cbffd8; }
512    footer { padding: 12px 14px 16px; display: grid; grid-template-columns: 1fr auto auto; gap: 9px; }
513    textarea { min-height: 46px; resize: vertical; }
514    @media (max-width: 620px) { footer, .bar { grid-template-columns: 1fr; } header { display: block; } }
515  </style>
516</head>
517<body>
518  <main>
519    <header>
520      <div>
521        <h1>aev-bridge</h1>
522        <div class="tag">HTTP + SSE bridge to aev-web</div>
523      </div>
524      <div id="status" class="tag">offline</div>
525    </header>
526    <section class="bar">
527      <input id="token" type="password" autocomplete="off" placeholder="Bridge token">
528      <button id="start">Start Session</button>
529    </section>
530    <pre id="screen"></pre>
531    <footer>
532      <textarea id="input" placeholder="Command, then Enter"></textarea>
533      <button id="send">Send</button>
534      <button id="ctrlc" class="secondary">Ctrl-C</button>
535    </footer>
536  </main>
537  <script>
538    const statusEl = document.querySelector('#status');
539    const screenEl = document.querySelector('#screen');
540    const tokenEl = document.querySelector('#token');
541    const inputEl = document.querySelector('#input');
542    let sessionId = null;
543    let events = null;
544
545    tokenEl.value = new URLSearchParams(location.search).get('token') || localStorage.getItem('aevBridgeToken') || '';
546
547    function token() {
548      const value = tokenEl.value.trim();
549      if (value) localStorage.setItem('aevBridgeToken', value);
550      return value;
551    }
552
553    async function api(path, options = {}) {
554      const headers = Object.assign({ 'Authorization': `Bearer ${token()}` }, options.headers || {});
555      if (options.body) headers['Content-Type'] = 'application/json';
556      const response = await fetch(`/api${path}`, Object.assign({}, options, { headers }));
557      const body = await response.json().catch(() => ({}));
558      if (!response.ok) throw new Error(body.error || response.statusText);
559      return body;
560    }
561
562    function connectEvents() {
563      if (events) events.close();
564      events = new EventSource(`/api/sessions/${sessionId}/events?token=${encodeURIComponent(token())}`);
565      events.addEventListener('output', event => {
566        const output = JSON.parse(event.data);
567        if (output.dropped) screenEl.textContent += '\n[bridge: output truncated]\n';
568        if (output.data) {
569          screenEl.textContent += output.data;
570          screenEl.scrollTop = screenEl.scrollHeight;
571        }
572        if (!output.alive) statusEl.textContent = `session ${sessionId} exited`;
573      });
574      events.addEventListener('error', () => { statusEl.textContent = 'event stream disconnected'; });
575    }
576
577    async function start() {
578      if (!token()) {
579        statusEl.textContent = 'token required';
580        return;
581      }
582      const result = await api('/sessions', { method: 'POST', body: JSON.stringify({}) });
583      sessionId = result.session.id;
584      screenEl.textContent = '';
585      statusEl.textContent = `session ${sessionId}`;
586      connectEvents();
587    }
588
589    async function send(data) {
590      if (!sessionId) await start();
591      if (!sessionId || !data) return;
592      await api(`/sessions/${sessionId}/input`, { method: 'POST', body: JSON.stringify({ data }) });
593    }
594
595    document.querySelector('#start').addEventListener('click', () => start().catch(error => statusEl.textContent = error.message));
596    document.querySelector('#send').addEventListener('click', () => {
597      const data = inputEl.value;
598      inputEl.value = '';
599      send(data).catch(error => statusEl.textContent = error.message);
600    });
601    document.querySelector('#ctrlc').addEventListener('click', () => send('\u0003').catch(error => statusEl.textContent = error.message));
602    inputEl.addEventListener('keydown', event => {
603      if (event.key === 'Enter' && !event.shiftKey) {
604        event.preventDefault();
605        const data = `${inputEl.value}\r`;
606        inputEl.value = '';
607        send(data).catch(error => statusEl.textContent = error.message);
608      }
609    });
610  </script>
611</body>
612</html>"#;
613
614#[cfg(test)]
615mod tests {
616    use super::*;
617    use axum::http::HeaderValue;
618    use std::ffi::CStr;
619
620    #[test]
621    fn manifest_matches_plugin_contract() {
622        let manifest = plugin_manifest();
623
624        assert_eq!(manifest.id, "aev-bridge");
625        assert_eq!(manifest.package, "aev-bridge");
626        assert_eq!(manifest.entrypoint, "aev_plugin_init");
627        assert_eq!(manifest.protocol, "aev.plugin.v1");
628        assert!(manifest.capabilities.contains(&"sse"));
629    }
630
631    #[test]
632    fn ffi_entrypoint_returns_static_manifest_json() {
633        assert!(!aev_plugin_init().is_null());
634        let json = unsafe { CStr::from_ptr(aev_plugin_init()) }
635            .to_str()
636            .expect("plugin manifest is utf-8");
637        assert_eq!(json, PLUGIN_INIT_JSON);
638    }
639
640    #[test]
641    fn auth_accepts_bearer_or_query_token() {
642        let mut headers = HeaderMap::new();
643        let uri = Uri::from_static("/api/sessions?token=secret");
644        assert!(request_token_matches(&headers, &uri, "secret"));
645
646        headers.insert(
647            header::AUTHORIZATION,
648            HeaderValue::from_static("Bearer secret"),
649        );
650        let uri = Uri::from_static("/api/sessions");
651        assert!(request_token_matches(&headers, &uri, "secret"));
652        assert!(!request_token_matches(&headers, &uri, "different"));
653    }
654
655    #[test]
656    fn upstream_base_url_is_normalized() {
657        assert_eq!(
658            normalize_upstream(" http://127.0.0.1:8765/// "),
659            "http://127.0.0.1:8765"
660        );
661    }
662
663    #[test]
664    fn router_constructs_with_capture_syntax() {
665        let config = ServerConfig {
666            bind: DEFAULT_BIND.parse().expect("bind parses"),
667            token: "bridge".to_string(),
668            upstream: DEFAULT_UPSTREAM.to_string(),
669            upstream_token: "upstream".to_string(),
670            public_url: None,
671            poll_ms: 300,
672        };
673        let state = AppState::new(config).expect("state builds");
674        let _ = router(state);
675    }
676}
aev_bridge/lib.rs

aev_bridge/
lib.rs
