use anyhow::Context;
use std::net::SocketAddr;
use std::{io, path::PathBuf};
use moq_auth::serve::{Keys, Policy};
use moq_auth::{Algorithm, Pattern, Permissions};
#[derive(usage::Args, Clone, Debug)]
#[usage(unknown_flags = "error", args_override_self = false)]
pub struct Args {
#[usage(subcommand)]
command: Command,
}
impl Args {
pub async fn run(self) -> anyhow::Result<()> {
match self.command {
Command::Serve(serve) => serve.run().await?,
Command::Generate {
algorithm,
id,
out,
out_dir,
public,
public_dir,
root,
publish,
subscribe,
} => {
let id = match id {
Some(id) => moq_auth::KeyId::decode(&id)?,
None => moq_auth::KeyId::random(),
};
let mut key = moq_auth::Key::generate(algorithm, Some(id.clone()))?;
if !publish.is_empty() || !subscribe.is_empty() {
key = key.with_scope(moq_auth::Scope {
root,
publish: publish.into_iter().collect(),
subscribe: subscribe.into_iter().collect(),
})?;
}
let public_to_stdout = public.as_deref().is_some_and(is_dash);
let private_to_stdout = out_dir.is_none() && out.as_deref().is_none_or(is_dash);
if public_to_stdout && private_to_stdout {
anyhow::bail!(
"cannot write both keys to stdout; use --out/--public with a file path, or --out-dir/--public-dir"
);
}
if let Some(dir) = public_dir {
let path = dir.join(format!("{id}.jwk"));
write_key(&key.to_public()?, &path)?;
} else if let Some(path) = public {
write_key(&key.to_public()?, &path)?;
}
if let Some(dir) = out_dir {
let path = dir.join(format!("{id}.jwk"));
write_key(&key, &path)?;
} else if let Some(path) = out {
write_key(&key, &path)?;
} else {
let encoded = key.to_str()?;
println!("{encoded}");
}
}
Command::Sign {
key,
root,
publish,
subscribe,
expires,
issued,
} => {
let key = read_key(&key)?;
let payload = moq_auth::Claims::default()
.with_root(root)
.with_publish(publish)
.with_subscribe(subscribe)
.with_expires(expires.map(Into::into))
.with_issued(issued.map(Into::into));
let token = key.sign(&payload)?;
println!("{token}");
}
Command::Verify { key, token } => {
if is_dash(&key) && is_dash(&token) {
anyhow::bail!("--key and --in cannot both read from stdin");
}
let key = read_key(&key)?;
let token = read_token(&token)?;
let payload = key.verify(&token)?;
println!("{payload:#?}");
}
Command::Sessions(filter) => filter.get("/sessions").await?,
Command::Revalidate(filter) => filter.post("/sessions/revalidate").await?,
}
Ok(())
}
}
#[derive(usage::Subcommands, Clone, Debug)]
enum Command {
Generate {
#[usage(long, default = "HS256")]
algorithm: Algorithm,
#[usage(long)]
id: Option<String>,
#[usage(long, value_hint = usage::ValueHint::FilePath, extensions("jwk", "json"))]
out: Option<PathBuf>,
#[usage(long, conflicts = "--out", value_hint = usage::ValueHint::DirPath)]
out_dir: Option<PathBuf>,
#[usage(long, value_hint = usage::ValueHint::FilePath, extensions("jwk", "json"))]
public: Option<PathBuf>,
#[usage(long, conflicts = "--public", value_hint = usage::ValueHint::DirPath)]
public_dir: Option<PathBuf>,
#[usage(long, default = "")]
root: String,
#[usage(long)]
publish: Vec<Pattern>,
#[usage(long)]
subscribe: Vec<Pattern>,
},
Sign {
#[usage(long, value_hint = usage::ValueHint::FilePath, extensions("jwk", "json"))]
key: PathBuf,
#[usage(long, default = "")]
root: String,
#[usage(long)]
publish: Vec<Pattern>,
#[usage(long)]
subscribe: Vec<Pattern>,
#[usage(long)]
expires: Option<UnixTimestamp>,
#[usage(long)]
issued: Option<UnixTimestamp>,
},
Serve(Serve),
Sessions(SessionFilter),
Revalidate(SessionFilter),
Verify {
#[usage(long, value_hint = usage::ValueHint::FilePath, extensions("jwk", "json"))]
key: PathBuf,
#[usage(long = "in", default = "-", value_hint = usage::ValueHint::FilePath)]
token: PathBuf,
},
}
#[derive(usage::Args, Clone, Debug)]
#[usage(unknown_flags = "error", args_override_self = false)]
pub struct SessionFilter {
#[usage(long)]
internal_url: url::Url,
#[usage(long)]
id: Option<String>,
#[usage(long)]
path: Option<String>,
#[usage(long)]
remote: Option<String>,
#[usage(long)]
transport: Option<String>,
#[usage(long)]
node: Option<String>,
#[usage(long)]
local: Option<String>,
#[usage(long)]
server_name: Option<String>,
#[usage(long)]
alpn: Option<String>,
#[usage(long)]
role: Option<String>,
#[usage(long)]
tls_name: Option<String>,
#[usage(long)]
tls_fingerprint: Option<String>,
#[usage(long)]
tls_issuer: Option<String>,
}
impl SessionFilter {
fn query(&self) -> Vec<(&'static str, &str)> {
let mut query = Vec::new();
if let Some(value) = &self.id {
query.push(("id", value.as_str()));
}
if let Some(value) = &self.path {
query.push(("path", value.as_str()));
}
if let Some(value) = &self.remote {
query.push(("remote", value.as_str()));
}
if let Some(value) = &self.transport {
query.push(("transport", value.as_str()));
}
if let Some(value) = &self.node {
query.push(("node", value.as_str()));
}
if let Some(value) = &self.local {
query.push(("local", value.as_str()));
}
if let Some(value) = &self.server_name {
query.push(("server_name", value.as_str()));
}
if let Some(value) = &self.alpn {
query.push(("alpn", value.as_str()));
}
if let Some(value) = &self.role {
query.push(("role", value.as_str()));
}
if let Some(value) = &self.tls_name {
query.push(("tls.name", value.as_str()));
}
if let Some(value) = &self.tls_fingerprint {
query.push(("tls.fingerprint", value.as_str()));
}
if let Some(value) = &self.tls_issuer {
query.push(("tls.issuer", value.as_str()));
}
query
}
fn url(&self, path: &str) -> url::Url {
let mut url = self.internal_url.clone();
let base = url.path().trim_end_matches('/');
url.set_path(&format!("{base}{path}"));
let query = self.query();
if !query.is_empty() {
url.set_query(Some(
&url::form_urlencoded::Serializer::new(String::new())
.extend_pairs(query)
.finish(),
));
}
url
}
async fn get(&self, path: &str) -> anyhow::Result<()> {
self.send(reqwest::Method::GET, path).await
}
async fn post(&self, path: &str) -> anyhow::Result<()> {
self.send(reqwest::Method::POST, path).await
}
async fn send(&self, method: reqwest::Method, path: &str) -> anyhow::Result<()> {
let response = reqwest::Client::new()
.request(method, self.url(path))
.send()
.await
.with_context(|| format!("failed to reach {}", self.internal_url))?;
let status = response.status();
let body = response.text().await.context("failed to read the reply")?;
if !status.is_success() {
anyhow::bail!("{status}: {body}");
}
match serde_json::from_str::<serde_json::Value>(&body) {
Ok(value) => println!("{}", serde_json::to_string_pretty(&value)?),
Err(_) => println!("{body}"),
}
Ok(())
}
}
#[derive(usage::Args, Clone, Debug)]
#[usage(unknown_flags = "error", args_override_self = false)]
pub struct Serve {
#[usage(long, default = "127.0.0.1:4440")]
listen: String,
#[usage(long)]
listen_public: bool,
#[usage(long, conflicts = "--key-dir", value_hint = usage::ValueHint::FilePath, extensions("jwk", "json"))]
key: Option<PathBuf>,
#[usage(long, conflicts = "--key", value_hint = usage::ValueHint::DirPath)]
key_dir: Option<PathBuf>,
#[usage(long)]
public_publish: Vec<Pattern>,
#[usage(long)]
public_subscribe: Vec<Pattern>,
#[usage(long)]
mtls_publish: Vec<Pattern>,
#[usage(long)]
mtls_subscribe: Vec<Pattern>,
#[usage(long)]
tier: Option<String>,
#[usage(long, default = "1m")]
revalidate: crate::duration::Duration,
#[usage(long, default = "1d")]
expires: crate::duration::Duration,
#[usage(long)]
limit_token: Option<usize>,
#[usage(long)]
limit_remote: Option<usize>,
}
impl Serve {
fn policy(&self) -> anyhow::Result<Policy> {
let revalidate = self.revalidate.into_std();
if revalidate.is_zero() {
anyhow::bail!("--revalidate must be longer than 0s; every client would re-check in a tight loop");
}
let rules = |publish: &[Pattern], subscribe: &[Pattern]| {
Permissions::new(publish.iter().cloned().collect(), subscribe.iter().cloned().collect())
};
let mut policy = Policy::default();
policy.keys = match (&self.key, &self.key_dir) {
(Some(key), _) => Some(Keys::File(key.clone())),
(None, Some(dir)) => Some(Keys::Dir(dir.clone())),
(None, None) => None,
};
policy.public = rules(&self.public_publish, &self.public_subscribe);
policy.mtls = rules(&self.mtls_publish, &self.mtls_subscribe);
policy.tier = self.tier.clone();
policy.revalidate = revalidate;
policy.expires = self.expires.into_std();
policy.limits.token = self.limit_token;
policy.limits.remote = self.limit_remote;
Ok(policy)
}
fn listener(&self) -> anyhow::Result<Listen> {
if let Some(path) = self.listen.strip_prefix("unix:") {
return Ok(Listen::Unix(path.into()));
}
let addr: SocketAddr = self
.listen
.parse()
.with_context(|| format!("--listen expects a socket address or unix:<path>, got {}", self.listen))?;
if !addr.ip().is_loopback() && !self.listen_public {
anyhow::bail!(
"--listen {addr} is not a loopback address; the server has no authentication of its own, so pass --listen-public to expose it on purpose"
);
}
Ok(Listen::Tcp(addr))
}
async fn run(self) -> anyhow::Result<()> {
let listen = self.listener()?;
let server = moq_auth::serve::Server::new(self.policy()?);
match listen {
Listen::Tcp(addr) => {
let listener = tokio::net::TcpListener::bind(addr)
.await
.with_context(|| format!("failed to bind {addr}"))?;
tracing::info!(%addr, "auth server listening");
server.serve(listener).await?;
}
#[cfg(unix)]
Listen::Unix(path) => {
unlink_socket(&path)?;
let listener = tokio::net::UnixListener::bind(&path)
.with_context(|| format!("failed to bind {}", path.display()))?;
tracing::info!(path = %path.display(), "auth server listening");
server.serve_unix(listener).await?;
}
#[cfg(not(unix))]
Listen::Unix(path) => anyhow::bail!("unix sockets are not supported here: {}", path.display()),
}
Ok(())
}
}
#[derive(Debug)]
enum Listen {
Tcp(SocketAddr),
Unix(PathBuf),
}
#[cfg(unix)]
fn unlink_socket(path: &std::path::Path) -> anyhow::Result<()> {
use std::os::unix::fs::FileTypeExt;
match std::fs::symlink_metadata(path) {
Ok(meta) if meta.file_type().is_socket() => {
std::fs::remove_file(path).with_context(|| format!("failed to remove the stale socket {}", path.display()))
}
Ok(_) => anyhow::bail!("{} exists and is not a socket; refusing to replace it", path.display()),
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(err).with_context(|| format!("failed to inspect {}", path.display())),
}
}
fn is_dash(path: &std::path::Path) -> bool {
path == std::path::Path::new("-")
}
fn write_key(key: &moq_auth::Key, path: &std::path::Path) -> anyhow::Result<()> {
if is_dash(path) {
println!("{}", key.to_str()?);
Ok(())
} else {
key.to_file(path)
.with_context(|| format!("failed to write key to {}", path.display()))
}
}
fn read_key(path: &std::path::Path) -> anyhow::Result<moq_auth::Key> {
if is_dash(path) {
let contents = io::read_to_string(io::stdin())?;
moq_auth::Key::from_str(contents.trim()).context("failed to parse key from stdin")
} else {
moq_auth::Key::from_file(path).with_context(|| format!("failed to read key from {}", path.display()))
}
}
fn read_token(path: &std::path::Path) -> anyhow::Result<String> {
let raw = if is_dash(path) {
io::read_to_string(io::stdin())?
} else {
std::fs::read_to_string(path).with_context(|| format!("failed to read token from {}", path.display()))?
};
Ok(raw.trim().to_string())
}
#[derive(Clone, Copy, Debug)]
struct UnixTimestamp(std::time::SystemTime);
impl From<UnixTimestamp> for std::time::SystemTime {
fn from(value: UnixTimestamp) -> Self {
value.0
}
}
impl std::str::FromStr for UnixTimestamp {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let timestamp = s.parse::<i64>().context("expected unix timestamp")?;
let timestamp = timestamp.try_into().context("timestamp out of range")?;
let value = std::time::SystemTime::UNIX_EPOCH
.checked_add(std::time::Duration::from_secs(timestamp))
.context("timestamp out of range")?;
Ok(Self(value))
}
}
#[cfg(test)]
mod tests {
use super::*;
async fn run(args: &[&str]) -> anyhow::Result<()> {
parse(args)?.run().await
}
fn parse(args: &[&str]) -> anyhow::Result<Args> {
let mut cli =
crate::args::Invocation::try_parse_from(args.iter().copied()).map_err(|err| anyhow::anyhow!("{err}"))?;
match cli.stages.remove(0) {
crate::args::Command::Auth(auth) => Ok(auth),
other => anyhow::bail!("parsed something other than `moq auth`: {}", other.name()),
}
}
fn serve(args: &[&str]) -> Serve {
match parse(args).unwrap().command {
Command::Serve(serve) => serve,
other => panic!("expected serve, got {other:?}"),
}
}
#[test]
fn serve_refuses_a_public_bind_without_saying_so() {
let err = serve(&["moq", "auth", "serve", "--listen", "0.0.0.0:4440"])
.listener()
.unwrap_err();
assert!(err.to_string().contains("--listen-public"), "{err}");
assert!(matches!(
serve(&["moq", "auth", "serve", "--listen", "0.0.0.0:4440", "--listen-public"]).listener(),
Ok(Listen::Tcp(_))
));
assert!(
matches!(serve(&["moq", "auth", "serve"]).listener(), Ok(Listen::Tcp(addr)) if addr.ip().is_loopback())
);
assert!(matches!(
serve(&["moq", "auth", "serve", "--listen", "unix:/run/moq-auth.sock"]).listener(),
Ok(Listen::Unix(path)) if path == std::path::Path::new("/run/moq-auth.sock")
));
assert!(
serve(&["moq", "auth", "serve", "--listen", "nowhere"])
.listener()
.is_err()
);
}
#[test]
fn serve_flags_become_the_policy() {
let policy = serve(&[
"moq",
"auth",
"serve",
"--key-dir",
"/keys",
"--public-subscribe",
"anon/**",
"--mtls-publish",
"**",
"--mtls-subscribe",
"**",
"--tier",
"internal",
"--revalidate",
"30s",
"--expires",
"2h",
"--limit-token",
"3",
"--limit-remote",
"8",
])
.policy()
.unwrap();
assert!(matches!(&policy.keys, Some(Keys::Dir(dir)) if dir == std::path::Path::new("/keys")));
assert_eq!(
policy.public.subscribe,
["anon/**".parse().unwrap()].into_iter().collect()
);
assert!(policy.public.publish.is_empty());
assert_eq!(policy.mtls.publish, ["**".parse().unwrap()].into_iter().collect());
assert_eq!(policy.tier.as_deref(), Some("internal"));
assert_eq!(policy.revalidate, std::time::Duration::from_secs(30));
assert_eq!(policy.expires, std::time::Duration::from_secs(7200));
assert_eq!(policy.limits.token, Some(3));
assert_eq!(policy.limits.remote, Some(8));
let bare = serve(&["moq", "auth", "serve"]).policy().unwrap();
assert!(bare.keys.is_none());
assert!(bare.public.is_empty() && bare.mtls.is_empty());
assert_eq!(bare.revalidate, std::time::Duration::from_secs(60));
assert_eq!(bare.expires, std::time::Duration::from_secs(86400));
let err = serve(&["moq", "auth", "serve", "--revalidate", "0s"])
.policy()
.unwrap_err();
assert!(err.to_string().contains("--revalidate"), "{err}");
}
#[cfg(unix)]
#[test]
fn unlink_socket_leaves_anything_but_a_socket_alone() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("config");
std::fs::write(&file, "keep me").unwrap();
let err = unlink_socket(&file).unwrap_err();
assert!(err.to_string().contains("not a socket"), "{err}");
assert_eq!(std::fs::read_to_string(&file).unwrap(), "keep me");
unlink_socket(&dir.path().join("missing")).unwrap();
let stale = dir.path().join("auth.sock");
drop(std::os::unix::net::UnixListener::bind(&stale).unwrap());
unlink_socket(&stale).unwrap();
assert!(!stale.exists());
}
#[tokio::test]
async fn generate_writes_a_usable_keypair() {
let dir = tempfile::tempdir().unwrap();
let private = dir.path().join("private.jwk");
let public = dir.path().join("public.jwk");
run(&[
"moq",
"auth",
"generate",
"--algorithm",
"ES256",
"--out",
private.to_str().unwrap(),
"--public",
public.to_str().unwrap(),
])
.await
.unwrap();
run(&[
"moq",
"auth",
"sign",
"--key",
private.to_str().unwrap(),
"--root",
"demo",
"--publish",
"alice/**",
"--issued",
"1700000000",
"--expires",
"4102444800",
])
.await
.unwrap();
let token = moq_auth::Key::from_file(&private)
.unwrap()
.sign(
&moq_auth::Claims::default()
.with_root("demo")
.with_publish(["alice/**".parse().unwrap()]),
)
.unwrap();
let path = dir.path().join("alice.jwt");
std::fs::write(&path, &token).unwrap();
run(&[
"moq",
"auth",
"verify",
"--key",
public.to_str().unwrap(),
"--in",
path.to_str().unwrap(),
])
.await
.unwrap();
}
#[tokio::test]
async fn generate_to_a_directory_names_the_file_after_the_kid() {
let dir = tempfile::tempdir().unwrap();
run(&["moq", "auth", "generate", "--out-dir", dir.path().to_str().unwrap()])
.await
.unwrap();
let written: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.map(|e| e.unwrap().path())
.collect();
assert_eq!(written.len(), 1, "expected exactly one key, got {written:?}");
let key = moq_auth::Key::from_file(&written[0]).unwrap();
let kid = key.kid.as_ref().expect("generate assigns a kid");
assert_eq!(written[0].file_name().unwrap().to_str().unwrap(), format!("{kid}.jwk"));
}
#[tokio::test]
async fn both_keys_to_stdout_is_rejected() {
let err = run(&["moq", "auth", "generate", "--algorithm", "ES256", "--public", "-"])
.await
.unwrap_err();
assert!(err.to_string().contains("cannot write both keys to stdout"), "{err}");
}
#[tokio::test]
async fn both_inputs_from_stdin_is_rejected() {
let err = run(&["moq", "auth", "verify", "--key", "-", "--in", "-"])
.await
.unwrap_err();
assert!(err.to_string().contains("cannot both read from stdin"), "{err}");
}
#[test]
fn revalidate_and_sessions_flags_parse() {
let args = parse(&[
"moq",
"auth",
"revalidate",
"--internal-url",
"http://127.0.0.1:9101",
"--id",
"abc",
"--path",
"demo/**",
"--remote",
"203.0.113.0/24",
"--tls-name",
"edge0",
])
.unwrap();
match args.command {
Command::Revalidate(filter) => {
assert_eq!(filter.id.as_deref(), Some("abc"));
assert_eq!(filter.path.as_deref(), Some("demo/**"));
assert_eq!(filter.remote.as_deref(), Some("203.0.113.0/24"));
assert_eq!(filter.tls_name.as_deref(), Some("edge0"));
}
other => panic!("expected revalidate, got {other:?}"),
}
assert!(matches!(
parse(&["moq", "auth", "sessions", "--internal-url", "http://127.0.0.1:9101"])
.unwrap()
.command,
Command::Sessions(_)
));
}
#[tokio::test]
async fn revalidate_posts_against_the_internal_listener() {
let sessions = moq_relay::session::Registry::new();
let mut request = moq_auth::Request::new("relay-1", moq_auth::Transport::Quic, "/demo/room");
request.id = "abc".into();
request.remote = Some("203.0.113.9:4433".parse().unwrap());
request.query = Some("jwt=secret".into());
let _reg = sessions.register(request);
let mut internal_config = moq_relay::internal::Config::default();
internal_config.listen = Some("127.0.0.1:0".parse().unwrap());
let internal =
moq_relay::internal::Internal::new(internal_config, moq_tokio::moq_net::stats::Registry::disabled())
.with_sessions(sessions)
.bind()
.expect("bind internal listener");
let listen = internal.addr().expect("internal listener is configured");
tokio::spawn(async move {
let _ = internal.run().await;
});
let url = format!("http://{listen}");
run(&["moq", "auth", "revalidate", "--internal-url", &url, "--id", "abc"])
.await
.expect("revalidate");
run(&["moq", "auth", "sessions", "--internal-url", &url, "--path", "demo/**"])
.await
.expect("sessions");
}
#[test]
fn timestamp_before_the_epoch_is_rejected() {
assert!("-1".parse::<UnixTimestamp>().is_err());
assert!("not-a-number".parse::<UnixTimestamp>().is_err());
}
#[test]
fn timestamp_at_the_maximum_does_not_panic() {
let _ = i64::MAX.to_string().parse::<UnixTimestamp>();
}
}