deploy slice A: engine hygiene — TCP_NODELAY + WS pings + trunk config #24
@@ -36,7 +36,24 @@ async fn main() {
|
||||
// Placeholder sender so we can build an AppState; the real sender comes
|
||||
// from the spawned MediaThread below.
|
||||
let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
|
||||
let app_state = AppState::new(placeholder_tx, default_tap_url);
|
||||
// Deploy slice A §5.3: config::twilio_credentials was parsed-but-
|
||||
// never-called since slice-5 — wired into startup validation here.
|
||||
// Partial or malformed RUTSTER_TWILIO_* config now fails the boot
|
||||
// loudly instead of silently running WebRTC-only. (The credentials
|
||||
// themselves stay green-zone: only webhook_base is threaded into the
|
||||
// routes; TwilioCredentials' Debug redacts auth_token — ADR-0009.)
|
||||
let twilio_credentials = rutster::config::twilio_credentials(
|
||||
std::env::var("RUTSTER_TWILIO_ACCOUNT_SID").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_AUTH_TOKEN").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_MEDIA_BIND").ok(),
|
||||
std::env::var("RUTSTER_TWILIO_WEBHOOK_BASE").ok(),
|
||||
)
|
||||
.expect("RUTSTER_TWILIO_* env config invalid");
|
||||
if let Some(creds) = twilio_credentials.as_ref() {
|
||||
info!(webhook_base = %creds.webhook_base, "trunk configured");
|
||||
}
|
||||
let app_state = AppState::new(placeholder_tx, default_tap_url)
|
||||
.with_trunk_webhook_base(twilio_credentials.as_ref().map(|c| c.webhook_base.clone()));
|
||||
|
||||
let media_cfg = rutster::config::media_address_config(
|
||||
std::env::var("RUTSTER_MEDIA_BIND_IP").ok(),
|
||||
|
||||
@@ -242,29 +242,76 @@ pub async fn originate_trunk_call() -> Response {
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// POST /v1/trunk/webhook -- Twilio's inbound-call signaling webhook
|
||||
/// receiver (spec §3.6 + §4.1 step 2). The handler responds with TwiML
|
||||
/// instructing Twilio to Media Streams against our /twilio/media-stream
|
||||
/// endpoint. The TwiML is:
|
||||
/// Derive the TwiML `<Stream>` URL from the operator's public webhook
|
||||
/// base (`RUTSTER_TWILIO_WEBHOOK_BASE`): `https://…` → `wss://…`,
|
||||
/// `http://…` → `ws://…` (dev loop), path fixed to the WS route's mount
|
||||
/// point.
|
||||
///
|
||||
/// ```xml
|
||||
/// <Response>
|
||||
/// <Connect>
|
||||
/// <Stream url="wss://{webhook_base}/twilio/media-stream" />
|
||||
/// </Connect>
|
||||
/// </Response>
|
||||
/// ```
|
||||
/// Authority-only by design: any path on the base is dropped, because
|
||||
/// `TwilioMediaStreamsServer::router` mounts the WS route absolutely at
|
||||
/// `/twilio/media-stream` — a path-prefixing edge must strip its prefix
|
||||
/// before the FOB, exactly like it must for every other route.
|
||||
fn stream_url_from_webhook_base(base: &url::Url) -> Result<url::Url, String> {
|
||||
let ws_scheme = match base.scheme() {
|
||||
"https" => "wss",
|
||||
"http" => "ws",
|
||||
other => {
|
||||
return Err(format!(
|
||||
"webhook base must be http(s); got scheme {other:?}"
|
||||
));
|
||||
}
|
||||
};
|
||||
let mut stream_url = base
|
||||
.join("/twilio/media-stream")
|
||||
.map_err(|e| format!("webhook base rejects a path join: {e}"))?;
|
||||
stream_url
|
||||
.set_scheme(ws_scheme)
|
||||
.map_err(|()| format!("cannot rewrite scheme to {ws_scheme}"))?;
|
||||
Ok(stream_url)
|
||||
}
|
||||
|
||||
/// POST /v1/trunk/webhook — Twilio's inbound-call signaling webhook
|
||||
/// receiver (deploy slice A §5.3). Responds with TwiML instructing Twilio
|
||||
/// to open a Media Streams fork against our `/twilio/media-stream` WS
|
||||
/// endpoint, at the PUBLIC address derived from
|
||||
/// `RUTSTER_TWILIO_WEBHOOK_BASE` — replacing the slice-5 placeholder's
|
||||
/// hardcoded `ws://127.0.0.1:8080`, which no CPaaS could ever dial.
|
||||
///
|
||||
/// Stub T5: emits the TwiML but the URL is hardcoded loopback for now.
|
||||
/// The full impl lands with the Step B refactor + dev-b's TwilioCredentials
|
||||
/// env parser (`webhook_base` from RUTSTER_THIRTY_WEBHOOK_BASE).
|
||||
pub async fn twilio_inbound_webhook() -> Response {
|
||||
let twiml = r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
/// 503 when the trunk is unconfigured: a webhook arriving without
|
||||
/// `RUTSTER_TWILIO_*` set is an operator misconfiguration (Twilio was
|
||||
/// pointed here, the binary wasn't told its public name) — failing the
|
||||
/// call loudly beats returning a loopback Stream URL that dies silently
|
||||
/// 10 s later. No XML escaping needed: the derived URL is
|
||||
/// scheme+authority+fixed-path, no query, and `url` forbids `"` in
|
||||
/// authorities.
|
||||
pub async fn twilio_inbound_webhook(State(state): State<AppState>) -> Response {
|
||||
let Some(base) = state.trunk_webhook_base.as_ref() else {
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"trunk not configured: set the RUTSTER_TWILIO_* env vars \
|
||||
(webhook received, but there is no RUTSTER_TWILIO_WEBHOOK_BASE \
|
||||
to derive the public Stream URL from)",
|
||||
)
|
||||
.into_response();
|
||||
};
|
||||
let stream_url = match stream_url_from_webhook_base(base) {
|
||||
Ok(u) => u,
|
||||
Err(e) => {
|
||||
// Defensive: config::twilio_credentials validated the URL at
|
||||
// startup, so this arm is unreachable in a correctly-booted
|
||||
// binary. 500, never a panic.
|
||||
tracing::error!(error = %e, "webhook base failed Stream-URL derivation");
|
||||
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
|
||||
}
|
||||
};
|
||||
let twiml = format!(
|
||||
r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Response>
|
||||
<Connect>
|
||||
<Stream url="ws://127.0.0.1:8080/twilio/media-stream" />
|
||||
<Stream url="{stream_url}" />
|
||||
</Connect>
|
||||
</Response>"#;
|
||||
</Response>"#
|
||||
);
|
||||
(
|
||||
StatusCode::OK,
|
||||
[(header::CONTENT_TYPE, "application/xml")],
|
||||
@@ -332,4 +379,70 @@ mod tests {
|
||||
.unwrap();
|
||||
assert_eq!(health.status(), StatusCode::OK);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_maps_https_to_wss_and_keeps_authority() {
|
||||
let base = url::Url::parse("https://pbx.example.com:8443").unwrap();
|
||||
let u = stream_url_from_webhook_base(&base).unwrap();
|
||||
assert_eq!(u.as_str(), "wss://pbx.example.com:8443/twilio/media-stream");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_maps_http_to_ws_for_dev() {
|
||||
let base = url::Url::parse("http://localhost:8080").unwrap();
|
||||
let u = stream_url_from_webhook_base(&base).unwrap();
|
||||
assert_eq!(u.as_str(), "ws://localhost:8080/twilio/media-stream");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_url_derivation_rejects_non_http_schemes() {
|
||||
let base = url::Url::parse("ftp://pbx.example.com").unwrap();
|
||||
let err = stream_url_from_webhook_base(&base).unwrap_err();
|
||||
assert!(err.contains("http"), "msg: {err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn webhook_derives_stream_url_from_configured_base() {
|
||||
let state = AppState::default()
|
||||
.with_trunk_webhook_base(Some(url::Url::parse("https://pbx.example.com").unwrap()));
|
||||
let app = router(state);
|
||||
let resp = app
|
||||
.oneshot(
|
||||
axum::http::Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/trunk/webhook")
|
||||
.body(axum::body::Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
let body = axum::body::to_bytes(resp.into_body(), 4096).await.unwrap();
|
||||
let twiml = String::from_utf8(body.to_vec()).unwrap();
|
||||
assert!(
|
||||
twiml.contains(r#"<Stream url="wss://pbx.example.com/twilio/media-stream" />"#),
|
||||
"twiml: {twiml}"
|
||||
);
|
||||
assert!(
|
||||
!twiml.contains("127.0.0.1"),
|
||||
"hardcoded loopback must be gone: {twiml}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn webhook_503_when_trunk_unconfigured() {
|
||||
// Default AppState: trunk_webhook_base = None (no RUTSTER_TWILIO_*).
|
||||
let app = router(AppState::default());
|
||||
let resp = app
|
||||
.oneshot(
|
||||
axum::http::Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/trunk/webhook")
|
||||
.body(axum::body::Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,6 +24,11 @@ use crate::media_thread::{MediaCmd, MediaThread};
|
||||
pub struct AppState {
|
||||
pub cmd_tx: mpsc::Sender<MediaCmd>,
|
||||
pub default_tap_url: url::Url,
|
||||
/// Public base URL the CPaaS calls back on
|
||||
/// (`RUTSTER_TWILIO_WEBHOOK_BASE`). `None` = trunk unconfigured — the
|
||||
/// webhook answers 503 instead of minting a TwiML Stream URL nobody
|
||||
/// can dial (deploy slice A §5.3).
|
||||
pub trunk_webhook_base: Option<url::Url>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
@@ -37,9 +42,18 @@ impl AppState {
|
||||
Self {
|
||||
cmd_tx,
|
||||
default_tap_url,
|
||||
trunk_webhook_base: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Thread the operator's public webhook base (from
|
||||
/// `config::twilio_credentials`) into the routes. Builder-style so the
|
||||
/// two-arg `new` keeps its many existing call sites.
|
||||
pub fn with_trunk_webhook_base(mut self, base: Option<url::Url>) -> Self {
|
||||
self.trunk_webhook_base = base;
|
||||
self
|
||||
}
|
||||
|
||||
/// Ask the media thread to create a fresh `RtcSession`, returning its id.
|
||||
pub async fn create_session(
|
||||
&self,
|
||||
|
||||
Reference in New Issue
Block a user