feat: finish Roku adapter dev-mode support

Add digest-auth sideload installs, debugger log capture, and live Roku
integration coverage so the full Roku milestone is validated on hardware.

src/adapters/mod.rs

@@ -11,6 +11,7 @@ pub mod roku;
 pub type Result<T> = std::result::Result<T, TvError>;
 
 /// A platform adapter capable of controlling one class of TVs.
+#[allow(async_fn_in_trait)]
 pub trait TvAdapter: Send + Sync {
     /// Discover candidate devices for this platform.
     async fn discover(&self) -> Result<Vec<DeviceInfo>>;

src/adapters/roku/mod.rs

@@ -1,16 +1,20 @@
 use std::{
     collections::{BTreeMap, BTreeSet},
+    env,
     net::IpAddr,
     time::Duration,
 };
 
 use chrono::Utc;
-use reqwest::{Client, StatusCode, Url};
+use reqwest::{Client, Method, StatusCode, Url, header::WWW_AUTHENTICATE, multipart};
 use roxmltree::Document;
 use tokio::{
+    io::AsyncReadExt,
+    net::TcpStream,
     net::UdpSocket,
     time::{Instant, timeout},
 };
+use uuid::Uuid;
 
 use super::{
     AppInfo, Device, DeviceInfo, DeviceState, PowerState, Result, TvAdapter, TvError, TvKey,
@@ -26,6 +30,9 @@ const ROKU_ECP_DISCOVERY_REQUEST: &str = concat!(
 );
 const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 5;
 const DEFAULT_DISCOVERY_TIMEOUT_SECS: u64 = 3;
+const DEFAULT_DEV_LOG_WINDOW_SECS: u64 = 3;
+const ROKU_DEV_WEB_PORT: u16 = 80;
+const ROKU_DEV_DEBUG_PORT: u16 = 8085;
 
 /// A Roku ECP adapter backed by SSDP and HTTP requests.
 #[derive(Debug, Clone)]
@@ -33,6 +40,7 @@ pub struct RokuAdapter {
     client: Client,
     request_timeout: Duration,
     discovery_timeout: Duration,
+    dev_log_window: Duration,
 }
 
 impl RokuAdapter {
@@ -42,6 +50,7 @@ impl RokuAdapter {
             client: Client::new(),
             request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
             discovery_timeout: Duration::from_secs(DEFAULT_DISCOVERY_TIMEOUT_SECS),
+            dev_log_window: Duration::from_secs(DEFAULT_DEV_LOG_WINDOW_SECS),
         }
     }
 
@@ -51,6 +60,7 @@ impl RokuAdapter {
             client: Client::new(),
             request_timeout,
             discovery_timeout,
+            dev_log_window: Duration::from_secs(DEFAULT_DEV_LOG_WINDOW_SECS),
         }
     }
 
@@ -161,6 +171,175 @@ impl RokuAdapter {
         let xml = self.get_text(url).await?;
         parse_device_info(&xml)
     }
+
+    fn dev_base_url(device: &Device) -> Result<Url> {
+        let host = match device.address {
+            IpAddr::V4(address) => address.to_string(),
+            IpAddr::V6(address) => format!("[{address}]"),
+        };
+        Url::parse(&format!("http://{host}:{ROKU_DEV_WEB_PORT}/"))
+            .map_err(|error| TvError::Transport(format!("invalid Roku developer URL: {error}")))
+    }
+
+    async fn developer_credentials(&self) -> Result<(String, String)> {
+        let username =
+            env::var("TVCTL_ROKU_DEV_USERNAME").unwrap_or_else(|_| "rokudev".to_string());
+        let password = env::var("TVCTL_ROKU_DEV_PASSWORD").map_err(|_| {
+            TvError::Config("missing TVCTL_ROKU_DEV_PASSWORD for Roku developer mode".to_string())
+        })?;
+        Ok((username, password))
+    }
+
+    async fn digest_challenge(&self, url: &Url) -> Result<DigestChallenge> {
+        let response = self
+            .client
+            .get(url.clone())
+            .timeout(self.request_timeout)
+            .send()
+            .await
+            .map_err(|error| TvError::Transport(format!("GET {url} failed: {error}")))?;
+
+        if response.status() != StatusCode::UNAUTHORIZED {
+            return Err(TvError::Transport(format!(
+                "expected digest challenge from {url}, got status {}",
+                response.status()
+            )));
+        }
+
+        let header = response
+            .headers()
+            .get(WWW_AUTHENTICATE)
+            .and_then(|value| value.to_str().ok())
+            .ok_or_else(|| TvError::Transport(format!("missing digest challenge from {url}")))?;
+
+        DigestChallenge::parse(header)
+    }
+
+    fn digest_authorization(
+        challenge: &DigestChallenge,
+        username: &str,
+        password: &str,
+        method: &Method,
+        uri: &str,
+    ) -> String {
+        let cnonce = format!("{:x}", md5::compute(Uuid::new_v4().as_bytes()));
+        let nc = "00000001";
+        let ha1 = format!(
+            "{:x}",
+            md5::compute(format!("{username}:{}:{password}", challenge.realm))
+        );
+        let ha2 = format!("{:x}", md5::compute(format!("{}:{uri}", method.as_str())));
+        let response = format!(
+            "{:x}",
+            md5::compute(format!(
+                "{ha1}:{}:{nc}:{cnonce}:{}:{ha2}",
+                challenge.nonce, challenge.qop
+            ))
+        );
+
+        let mut parts = vec![
+            format!("username=\"{username}\""),
+            format!("realm=\"{}\"", challenge.realm),
+            format!("nonce=\"{}\"", challenge.nonce),
+            format!("uri=\"{uri}\""),
+            format!("response=\"{response}\""),
+            format!("qop={}", challenge.qop),
+            format!("nc={nc}"),
+            format!("cnonce=\"{cnonce}\""),
+        ];
+        if let Some(opaque) = &challenge.opaque {
+            parts.push(format!("opaque=\"{opaque}\""));
+        }
+        if let Some(algorithm) = &challenge.algorithm {
+            parts.push(format!("algorithm={algorithm}"));
+        }
+
+        format!("Digest {}", parts.join(", "))
+    }
+
+    async fn developer_post_multipart(
+        &self,
+        device: &Device,
+        path: &str,
+        form: multipart::Form,
+    ) -> Result<String> {
+        let (username, password) = self.developer_credentials().await?;
+        let base_url = Self::dev_base_url(device)?;
+        let url = Self::join_url(&base_url, path)?;
+        let challenge = self.digest_challenge(&url).await?;
+        let uri = format!("/{path}");
+        let authorization =
+            Self::digest_authorization(&challenge, &username, &password, &Method::POST, &uri);
+
+        let response = self
+            .client
+            .post(url.clone())
+            .timeout(self.request_timeout)
+            .header("Authorization", authorization)
+            .multipart(form)
+            .send()
+            .await
+            .map_err(|error| TvError::Transport(format!("POST {url} failed: {error}")))?;
+
+        let status = response.status();
+        let body = response
+            .text()
+            .await
+            .map_err(|error| TvError::Transport(format!("reading {url} failed: {error}")))?;
+
+        if status == StatusCode::UNAUTHORIZED {
+            return Err(TvError::Transport(
+                "invalid Roku developer credentials".to_string(),
+            ));
+        }
+
+        Ok(body)
+    }
+
+    async fn debugger_lines(&self, device: &Device) -> Result<Vec<String>> {
+        let address = (device.address, ROKU_DEV_DEBUG_PORT);
+        let mut stream = timeout(self.request_timeout, TcpStream::connect(address))
+            .await
+            .map_err(|_| {
+                TvError::Transport(format!(
+                    "connecting to Roku debugger at {} timed out",
+                    device.address
+                ))
+            })?
+            .map_err(|error| {
+                TvError::Transport(format!("connecting to Roku debugger failed: {error}"))
+            })?;
+
+        let deadline = Instant::now() + self.dev_log_window;
+        let mut buffer = vec![0_u8; 4096];
+        let mut output = Vec::new();
+
+        loop {
+            let now = Instant::now();
+            if now >= deadline {
+                break;
+            }
+
+            let remaining = deadline - now;
+            match timeout(remaining, stream.read(&mut buffer)).await {
+                Ok(Ok(0)) => break,
+                Ok(Ok(size)) => output.extend_from_slice(&buffer[..size]),
+                Ok(Err(error)) => {
+                    return Err(TvError::Transport(format!(
+                        "reading Roku debugger output failed: {error}"
+                    )));
+                }
+                Err(_) => break,
+            }
+        }
+
+        Ok(String::from_utf8_lossy(&output)
+            .lines()
+            .map(str::trim_end)
+            .filter(|line| !line.is_empty())
+            .map(ToString::to_string)
+            .collect())
+    }
 }
 
 impl Default for RokuAdapter {
@@ -285,6 +464,38 @@ impl TvAdapter for RokuAdapter {
     async fn dev_reload(&self, device: &Device) -> Result<()> {
         self.device_post(device, "launch/dev").await
     }
+
+    async fn dev_install(&self, device: &Device, zip: &[u8]) -> Result<()> {
+        let part = multipart::Part::bytes(zip.to_vec())
+            .file_name("tvctl-dev.zip")
+            .mime_str("application/zip")
+            .map_err(|error| {
+                TvError::Transport(format!("invalid Roku package MIME type: {error}"))
+            })?;
+        let form = multipart::Form::new()
+            .text("mysubmit", "Install")
+            .part("archive", part);
+
+        let body = self
+            .developer_post_multipart(device, "plugin_install", form)
+            .await?;
+
+        if body.contains("Install Success.") {
+            return Ok(());
+        }
+
+        if let Some(message) = extract_first_message(&body, "Install Failure:") {
+            return Err(TvError::Transport(message));
+        }
+
+        Err(TvError::Transport(
+            "Roku developer install did not report success".to_string(),
+        ))
+    }
+
+    async fn dev_logs(&self, device: &Device) -> Result<Vec<String>> {
+        self.debugger_lines(device).await
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -398,6 +609,18 @@ fn parse_ssdp_location(response: &str) -> Option<String> {
     })
 }
 
+fn extract_first_message(body: &str, prefix: &str) -> Option<String> {
+    body.lines()
+        .map(str::trim)
+        .find(|line| line.contains(prefix))
+        .map(|line| {
+            line.replace("<font color=\"red\">", "")
+                .replace("</font>", "")
+                .trim()
+                .to_string()
+        })
+}
+
 fn map_power_state(value: Option<&str>) -> PowerState {
     let Some(value) = value else {
         return PowerState::Unknown;
@@ -459,6 +682,46 @@ fn roku_key_paths(key: &TvKey) -> Result<Vec<String>> {
     Err(TvError::InvalidKey(path.to_string()))
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct DigestChallenge {
+    realm: String,
+    nonce: String,
+    qop: String,
+    opaque: Option<String>,
+    algorithm: Option<String>,
+}
+
+impl DigestChallenge {
+    fn parse(header: &str) -> Result<Self> {
+        let challenge = header
+            .strip_prefix("Digest ")
+            .ok_or_else(|| TvError::Transport(format!("unsupported auth challenge: {header}")))?;
+        let mut values = BTreeMap::new();
+        for part in challenge.split(',') {
+            let (key, value) = part
+                .trim()
+                .split_once('=')
+                .ok_or_else(|| TvError::Transport(format!("invalid digest challenge: {header}")))?;
+            values.insert(
+                key.trim().to_string(),
+                value.trim().trim_matches('"').to_string(),
+            );
+        }
+
+        Ok(Self {
+            realm: values
+                .remove("realm")
+                .ok_or_else(|| TvError::Transport("digest challenge missing realm".to_string()))?,
+            nonce: values
+                .remove("nonce")
+                .ok_or_else(|| TvError::Transport("digest challenge missing nonce".to_string()))?,
+            qop: values.remove("qop").unwrap_or_else(|| "auth".to_string()),
+            opaque: values.remove("opaque"),
+            algorithm: values.remove("algorithm"),
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -532,4 +795,14 @@ mod tests {
         let error = roku_key_paths(&TvKey::PowerOn).expect_err("PowerOn should not map");
         assert!(matches!(error, TvError::InvalidKey(key) if key == "power-on"));
     }
+
+    #[test]
+    fn parses_digest_challenge() {
+        let challenge =
+            DigestChallenge::parse("Digest qop=\"auth\", realm=\"rokudev\", nonce=\"1776173348\"")
+                .expect("challenge should parse");
+        assert_eq!(challenge.realm, "rokudev");
+        assert_eq!(challenge.nonce, "1776173348");
+        assert_eq!(challenge.qop, "auth");
+    }
 }

src/lib.rs

@@ -0,0 +1,6 @@
+#![allow(dead_code)]
+
+pub mod adapters;
+pub mod api;
+pub mod cli;
+pub mod daemon;

src/main.rs

@@ -1,10 +1,3 @@
-#![allow(dead_code)]
-
-mod adapters;
-mod api;
-mod cli;
-mod daemon;
-
 /// Launch the tvctl binary entry point.
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -17,5 +10,5 @@ async fn main() -> anyhow::Result<()> {
         .compact()
         .init();
 
-    cli::run().await
+    tvctl::cli::run().await
 }