supercode-opencode-frontend 0.5.52

//! Authenticated numeric-loopback HTTP host for the stock OpenCode adapter.

use std::fmt;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use base64::Engine as _;
use serde_json::Value;
use supercode_harness::SdkRuntime;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tokio::task::JoinSet;
use zeroize::Zeroize;

use crate::{OpenCodeAdapter, OpenCodeRequest, ResponseBody};

const CREDENTIAL_BYTES: usize = 32;
const MAX_HEADER_BYTES: usize = 64 * 1024;
const MAX_BODY_BYTES: usize = 1024 * 1024;
const MAX_EVENT_BYTES: usize = 16 * 1024 * 1024;
const MAX_CONCURRENT_CONNECTIONS: usize = 32;
const READ_TIMEOUT: Duration = Duration::from_secs(5);
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);

#[derive(Debug, thiserror::Error)]
pub enum CredentialError {
    #[error("operating system random source failed")]
    RandomSource,
}

/// Opaque stock-client password. It is non-cloneable, redacted, and zeroized;
/// the only production disclosure path installs it in one child environment.
pub struct OpenCodeClientCredential {
    password: [u8; CREDENTIAL_BYTES],
}

impl OpenCodeClientCredential {
    pub fn spawn_tokio_child(
        &self,
        command: &mut tokio::process::Command,
    ) -> std::io::Result<tokio::process::Child> {
        let mut password = encode_hex(&self.password);
        command.env("OPENCODE_SERVER_PASSWORD", &password);
        command.env("OPENCODE_SERVER_USERNAME", "opencode");
        let child = command.spawn();
        command.env_remove("OPENCODE_SERVER_PASSWORD");
        command.env_remove("OPENCODE_SERVER_USERNAME");
        password.zeroize();
        child
    }
}

impl fmt::Debug for OpenCodeClientCredential {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("OpenCodeClientCredential([REDACTED])")
    }
}

impl Drop for OpenCodeClientCredential {
    fn drop(&mut self) {
        self.password.zeroize();
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OpenCodeEndpointHealth {
    Ready,
    ShuttingDown,
    Stopped,
}

/// Removable HTTP endpoint around one already-authorized SDK runtime client.
pub struct OpenCodeEndpoint {
    adapter: Arc<OpenCodeAdapter>,
    credential_digest: [u8; 32],
}

impl OpenCodeEndpoint {
    pub fn new(
        runtime: Arc<dyn SdkRuntime>,
        runtime_id: impl Into<String>,
        workspace: impl Into<PathBuf>,
    ) -> Result<(Arc<Self>, OpenCodeClientCredential), CredentialError> {
        let mut password = [0_u8; CREDENTIAL_BYTES];
        getrandom::getrandom(&mut password).map_err(|_| CredentialError::RandomSource)?;
        let mut basic = format!("opencode:{}", encode_hex(&password));
        let credential_digest = *blake3::hash(basic.as_bytes()).as_bytes();
        basic.zeroize();
        Ok((
            Arc::new(Self {
                adapter: OpenCodeAdapter::new(runtime, runtime_id, workspace),
                credential_digest,
            }),
            OpenCodeClientCredential { password },
        ))
    }

    pub fn session_id(&self) -> &str {
        self.adapter.session_id()
    }

    /// Bind only numeric IPv4 loopback. Remote use is an operator-owned
    /// authenticated tunnel; there is no wildcard/arbitrary-address API.
    pub async fn bind(self: &Arc<Self>, port: u16) -> std::io::Result<OpenCodeServerHandle> {
        let listener =
            TcpListener::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port)).await?;
        let address = listener.local_addr()?;
        let (shutdown, shutdown_receiver) = watch::channel(false);
        let shutting_down = Arc::new(AtomicBool::new(false));
        let stopped = Arc::new(AtomicBool::new(false));
        let task = tokio::spawn(run_accept_loop(
            listener,
            self.clone(),
            address,
            shutdown_receiver,
            shutting_down.clone(),
            stopped.clone(),
        ));
        Ok(OpenCodeServerHandle {
            address,
            shutdown,
            shutting_down,
            stopped,
            task: Some(task),
        })
    }

    fn authenticate(&self, authorization: &str) -> bool {
        let Some(encoded) = authorization.strip_prefix("Basic ") else {
            return false;
        };
        let Ok(mut decoded) = base64::engine::general_purpose::STANDARD.decode(encoded) else {
            return false;
        };
        let digest = *blake3::hash(&decoded).as_bytes();
        decoded.zeroize();
        constant_time_eq(&digest, &self.credential_digest)
    }
}

pub struct OpenCodeServerHandle {
    address: SocketAddr,
    shutdown: watch::Sender<bool>,
    shutting_down: Arc<AtomicBool>,
    stopped: Arc<AtomicBool>,
    task: Option<tokio::task::JoinHandle<()>>,
}

impl OpenCodeServerHandle {
    pub fn address(&self) -> SocketAddr {
        self.address
    }

    pub fn url(&self) -> String {
        format!("http://{}", self.address)
    }

    pub fn health(&self) -> OpenCodeEndpointHealth {
        if self.stopped.load(Ordering::SeqCst) {
            OpenCodeEndpointHealth::Stopped
        } else if self.shutting_down.load(Ordering::SeqCst) {
            OpenCodeEndpointHealth::ShuttingDown
        } else {
            OpenCodeEndpointHealth::Ready
        }
    }

    pub async fn shutdown(mut self) {
        self.shutting_down.store(true, Ordering::SeqCst);
        self.shutdown.send_replace(true);
        if let Some(task) = self.task.take() {
            let _ = task.await;
        }
    }
}

impl Drop for OpenCodeServerHandle {
    fn drop(&mut self) {
        self.shutting_down.store(true, Ordering::SeqCst);
        self.shutdown.send_replace(true);
        if let Some(task) = self.task.take() {
            task.abort();
        }
    }
}

async fn run_accept_loop(
    listener: TcpListener,
    endpoint: Arc<OpenCodeEndpoint>,
    address: SocketAddr,
    mut shutdown: watch::Receiver<bool>,
    shutting_down: Arc<AtomicBool>,
    stopped: Arc<AtomicBool>,
) {
    let budget = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_CONNECTIONS));
    let mut connections = JoinSet::new();
    loop {
        tokio::select! {
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() {
                    break;
                }
            }
            _ = connections.join_next(), if !connections.is_empty() => {}
            accepted = listener.accept() => match accepted {
                Ok((stream, _)) => {
                    let Ok(permit) = budget.clone().try_acquire_owned() else {
                        drop(stream);
                        continue;
                    };
                    let endpoint = endpoint.clone();
                    let mut connection_shutdown = shutdown.clone();
                    connections.spawn(async move {
                        let _permit = permit;
                        tokio::select! {
                            _ = wait_for_shutdown(&mut connection_shutdown) => {}
                            _ = serve_connection(stream, endpoint, address) => {}
                        }
                    });
                }
                Err(_) => break,
            }
        }
    }
    connections.shutdown().await;
    shutting_down.store(true, Ordering::SeqCst);
    stopped.store(true, Ordering::SeqCst);
}

async fn wait_for_shutdown(shutdown: &mut watch::Receiver<bool>) {
    while !*shutdown.borrow() {
        if shutdown.changed().await.is_err() {
            break;
        }
    }
}

async fn serve_connection(
    mut stream: TcpStream,
    endpoint: Arc<OpenCodeEndpoint>,
    expected_address: SocketAddr,
) -> std::io::Result<()> {
    let request = match read_request(&mut stream, expected_address).await {
        Ok(request) => request,
        Err(error) => return write_error(&mut stream, error.status, error.message).await,
    };
    if !endpoint.authenticate(&request.authorization) {
        return write_error(&mut stream, 401, "authentication failed").await;
    }
    let body = if request.body.is_empty() {
        Value::Null
    } else {
        match serde_json::from_slice(&request.body) {
            Ok(body) => body,
            Err(_) => return write_error(&mut stream, 400, "invalid JSON body").await,
        }
    };
    // Runtime mutations outlive a frontend connection. If the endpoint is
    // shut down while a model turn is running, dropping this connection task
    // must not cancel or partially unwind the SDK-owned turn.
    let adapter = endpoint.adapter.clone();
    let operation = tokio::spawn(async move {
        adapter
            .handle(OpenCodeRequest {
                method: request.method,
                target: request.target,
                body,
            })
            .await
    });
    let response = operation.await.map_err(std::io::Error::other)?;
    match response.body {
        ResponseBody::Json(body) => write_json(&mut stream, response.status, &body).await,
        ResponseBody::EventStream(attachment) => {
            write_event_stream(&mut stream, endpoint.adapter.clone(), *attachment).await
        }
    }
}

struct WireRequest {
    method: String,
    target: String,
    authorization: String,
    body: Vec<u8>,
}

struct WireError {
    status: u16,
    message: &'static str,
}

async fn read_request(
    stream: &mut TcpStream,
    expected_address: SocketAddr,
) -> Result<WireRequest, WireError> {
    read_request_before(
        stream,
        expected_address,
        tokio::time::Instant::now() + READ_TIMEOUT,
    )
    .await
}

async fn read_request_before(
    stream: &mut TcpStream,
    expected_address: SocketAddr,
    deadline: tokio::time::Instant,
) -> Result<WireRequest, WireError> {
    let mut bytes = Vec::with_capacity(4096);
    let header_end = loop {
        if bytes.len() >= MAX_HEADER_BYTES {
            return Err(WireError {
                status: 431,
                message: "request headers too large",
            });
        }
        let mut chunk = [0_u8; 4096];
        let read = tokio::time::timeout_at(deadline, stream.read(&mut chunk))
            .await
            .map_err(|_| WireError {
                status: 408,
                message: "request read timeout",
            })?
            .map_err(|_| WireError {
                status: 400,
                message: "request read failed",
            })?;
        if read == 0 {
            return Err(WireError {
                status: 400,
                message: "incomplete request",
            });
        }
        bytes.extend_from_slice(&chunk[..read]);
        if let Some(index) = find_subslice(&bytes, b"\r\n\r\n") {
            if index + 4 > MAX_HEADER_BYTES {
                return Err(WireError {
                    status: 431,
                    message: "request headers too large",
                });
            }
            break index + 4;
        }
    };
    let header = std::str::from_utf8(&bytes[..header_end]).map_err(|_| WireError {
        status: 400,
        message: "request headers are not UTF-8",
    })?;
    let mut lines = header[..header.len() - 4].split("\r\n");
    let request_line = lines.next().ok_or(WireError {
        status: 400,
        message: "missing request line",
    })?;
    let fields = request_line.split(' ').collect::<Vec<_>>();
    if fields.len() != 3 || fields[2] != "HTTP/1.1" {
        return Err(WireError {
            status: 400,
            message: "invalid HTTP/1.1 request line",
        });
    }
    if !fields[1].starts_with('/') || fields[1].starts_with("//") {
        return Err(WireError {
            status: 400,
            message: "absolute-form target rejected",
        });
    }
    let method = fields[0].to_string();
    let target = fields[1].to_string();
    let mut host = None;
    let mut authorization = None;
    let mut content_length = None;
    for line in lines {
        let (name, value) = line.split_once(':').ok_or(WireError {
            status: 400,
            message: "invalid request header",
        })?;
        let name = name.to_ascii_lowercase();
        let value = value.trim();
        match name.as_str() {
            "host" if host.replace(value).is_some() => return Err(duplicate_security_header()),
            "authorization" if authorization.replace(value).is_some() => {
                return Err(duplicate_security_header())
            }
            "content-length" if content_length.replace(value).is_some() => {
                return Err(duplicate_security_header())
            }
            "transfer-encoding" | "origin" => {
                return Err(WireError {
                    status: 400,
                    message: "forbidden request header",
                })
            }
            _ => {}
        }
    }
    if host != Some(expected_address.to_string().as_str()) {
        return Err(WireError {
            status: 400,
            message: "invalid loopback Host",
        });
    }
    let authorization = authorization
        .ok_or(WireError {
            status: 401,
            message: "authentication failed",
        })?
        .to_string();
    let content_length = content_length
        .unwrap_or("0")
        .parse::<usize>()
        .map_err(|_| WireError {
            status: 400,
            message: "invalid content length",
        })?;
    if content_length > MAX_BODY_BYTES {
        return Err(WireError {
            status: 413,
            message: "request body too large",
        });
    }
    while bytes.len() - header_end < content_length {
        let remaining = content_length - (bytes.len() - header_end);
        let mut chunk = vec![0_u8; remaining.min(4096)];
        let read = tokio::time::timeout_at(deadline, stream.read(&mut chunk))
            .await
            .map_err(|_| WireError {
                status: 408,
                message: "request body timeout",
            })?
            .map_err(|_| WireError {
                status: 400,
                message: "request body read failed",
            })?;
        if read == 0 {
            return Err(WireError {
                status: 400,
                message: "incomplete request body",
            });
        }
        bytes.extend_from_slice(&chunk[..read]);
    }
    if bytes.len() - header_end != content_length {
        return Err(WireError {
            status: 400,
            message: "pipelined request bytes rejected",
        });
    }
    Ok(WireRequest {
        method,
        target,
        authorization,
        body: bytes[header_end..].to_vec(),
    })
}

async fn write_json(stream: &mut TcpStream, status: u16, body: &Value) -> std::io::Result<()> {
    let body = serde_json::to_vec(body).expect("JSON Value serializes");
    let reason = status_reason(status);
    let header = format!(
        "HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\nX-Content-Type-Options: nosniff\r\n\r\n",
        body.len()
    );
    stream.write_all(header.as_bytes()).await?;
    stream.write_all(&body).await?;
    stream.shutdown().await
}

async fn write_error(
    stream: &mut TcpStream,
    status: u16,
    message: &'static str,
) -> std::io::Result<()> {
    write_json(
        stream,
        status,
        &serde_json::json!({"name": "transport_error", "message": message}),
    )
    .await
}

async fn write_event_stream(
    stream: &mut TcpStream,
    adapter: Arc<OpenCodeAdapter>,
    mut attachment: supercode_harness::FrontendAttachment,
) -> std::io::Result<()> {
    stream
        .write_all(
            b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\nX-Accel-Buffering: no\r\n\r\n",
        )
        .await?;
    for event in adapter.initial_events(&attachment) {
        write_sse(stream, &event).await?;
    }
    let mut projection = adapter.event_projection(&attachment);
    loop {
        tokio::select! {
            event = attachment.next_event() => match event {
                Ok(event) => {
                    for projected in projection.project(&event) {
                        write_sse(stream, &projected).await?;
                    }
                }
                Err(_) => break,
            },
            _ = tokio::time::sleep(HEARTBEAT_INTERVAL) => {
                write_sse(stream, &serde_json::json!({"type": "server.heartbeat", "properties": {}})).await?;
            }
        }
    }
    stream.shutdown().await
}

async fn write_sse(stream: &mut TcpStream, event: &Value) -> std::io::Result<()> {
    let encoded = serde_json::to_vec(event).expect("JSON Value serializes");
    if encoded.len() > MAX_EVENT_BYTES {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "projected event exceeded byte limit",
        ));
    }
    stream.write_all(b"data: ").await?;
    stream.write_all(&encoded).await?;
    stream.write_all(b"\n\n").await?;
    stream.flush().await
}

fn duplicate_security_header() -> WireError {
    WireError {
        status: 400,
        message: "duplicate security header",
    }
}

fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack
        .windows(needle.len())
        .position(|window| window == needle)
}

fn constant_time_eq(left: &[u8; 32], right: &[u8; 32]) -> bool {
    left.iter()
        .zip(right)
        .fold(0_u8, |difference, (left, right)| {
            difference | (left ^ right)
        })
        == 0
}

fn encode_hex(secret: &[u8; CREDENTIAL_BYTES]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(CREDENTIAL_BYTES * 2);
    for byte in secret {
        encoded.push(HEX[(byte >> 4) as usize] as char);
        encoded.push(HEX[(byte & 0x0f) as usize] as char);
    }
    encoded
}

fn status_reason(status: u16) -> &'static str {
    match status {
        200 => "OK",
        400 => "Bad Request",
        401 => "Unauthorized",
        403 => "Forbidden",
        404 => "Not Found",
        408 => "Request Timeout",
        409 => "Conflict",
        413 => "Payload Too Large",
        431 => "Request Header Fields Too Large",
        500 => "Internal Server Error",
        _ => "Error",
    }
}