diff --git a/Cargo.lock b/Cargo.lock index c53e248..8e6c8f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1226,6 +1226,7 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "base64", "chrono", "clap", "figment", @@ -1240,6 +1241,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "sqlx", "thiserror 2.0.18", "tokio", diff --git a/Cargo.toml b/Cargo.toml index e5bd14a..237f044 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,3 +58,5 @@ url = "2" regex = "1" chrono = { version = "0.4", features = ["serde"] } rand = "0.8" +sha2 = "0.10" +base64 = "0.22" diff --git a/deploy/kindexr.example.yaml b/deploy/kindexr.example.yaml index 8bc38cf..dbcf704 100644 --- a/deploy/kindexr.example.yaml +++ b/deploy/kindexr.example.yaml @@ -46,6 +46,16 @@ health: method: "dht" # dht | tracker | both refresh_interval: "30m" +# Blossom binary bridge (BUD-01) +# When enabled, kindexr will: +# - Cache .torrent files linked via `url` tags in incoming NIP-35 events +# - Upload .torrent files to this server before publishing your own NIP-35 events +# - Serve cached files at /torrent/.torrent for *arr apps +# The server can be self-hosted (e.g. https://github.com/hzrd149/blossom-server). +blossom: + enabled: false + server: "https://blossom.example.com" + # Publisher — publish your own torrents as NIP-35 events publisher: enabled: false diff --git a/src/blossom/mod.rs b/src/blossom/mod.rs new file mode 100644 index 0000000..cc2e50c --- /dev/null +++ b/src/blossom/mod.rs @@ -0,0 +1,96 @@ +use anyhow::Context; +use base64::{engine::general_purpose::STANDARD as B64, Engine}; +use nostr::{EventBuilder, JsonUtil, Kind, Tag}; +use sha2::{Digest, Sha256}; +use tracing::debug; + +use crate::nostr::signer::Signer; + +/// Minimal BUD-01 Blossom client. +/// Fetch is unauthenticated (trusted public server). +/// Upload uses a kind 24242 Nostr auth event signed by the publisher identity. +pub struct BlossomClient { + server: String, // no trailing slash + http: reqwest::Client, +} + +#[derive(serde::Deserialize)] +struct BlobDescriptor { + url: String, +} + +impl BlossomClient { + pub fn new(server: &str) -> Self { + BlossomClient { + server: server.trim_end_matches('/').to_owned(), + http: reqwest::Client::new(), + } + } + + /// GET a blob by its full URL. Returns raw bytes. + pub async fn fetch(&self, url: &str) -> anyhow::Result> { + let resp = self.http.get(url).send().await.context("blossom fetch")?; + if !resp.status().is_success() { + anyhow::bail!("blossom fetch {}: HTTP {}", url, resp.status()); + } + Ok(resp.bytes().await.context("blossom fetch body")?.to_vec()) + } + + /// PUT a blob (BUD-01). Signs a kind 24242 auth event with the publisher key. + /// Returns the canonical URL the server assigned to the blob. + pub async fn upload(&self, data: &[u8], signer: &Signer) -> anyhow::Result { + let sha256 = hex::encode(Sha256::digest(data)); + let auth = make_upload_auth(&sha256, signer)?; + let auth_header = format!("Nostr {}", B64.encode(auth.as_json())); + + debug!(sha256 = %sha256, server = %self.server, "uploading blob to blossom"); + + let resp = self + .http + .put(format!("{}/upload", self.server)) + .header("Authorization", auth_header) + .header("Content-Type", "application/x-bittorrent") + .header("X-SHA-256", &sha256) + .body(data.to_vec()) + .send() + .await + .context("blossom upload")?; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + anyhow::bail!("blossom upload HTTP {status}: {body}"); + } + + let desc: BlobDescriptor = resp.json().await.context("blossom upload response")?; + Ok(desc.url) + } + + /// SHA-256 hex of a byte slice — useful for building the expected URL before upload. + pub fn sha256_hex(data: &[u8]) -> String { + hex::encode(Sha256::digest(data)) + } + + /// Derive the canonical Blossom URL for a known sha256 hash on this server. + pub fn blob_url(&self, sha256: &str) -> String { + format!("{}/{}", self.server, sha256) + } +} + +/// Build and sign a BUD-01 kind 24242 upload auth event (valid for 10 minutes). +fn make_upload_auth(sha256: &str, signer: &Signer) -> anyhow::Result { + let expiration = (now_secs() + 600).to_string(); + let builder = EventBuilder::new(Kind::Custom(24242), format!("Upload {sha256}")).tags([ + Tag::parse(["t", "upload"])?, + Tag::parse(["x", sha256])?, + Tag::parse(["expiration", &expiration])?, + ]); + signer.sign(builder) +} + +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} diff --git a/src/config.rs b/src/config.rs index 7e19f37..d0ba2fa 100644 --- a/src/config.rs +++ b/src/config.rs @@ -16,6 +16,7 @@ pub struct Config { pub tmdb: TmdbConfig, pub health: HealthConfig, pub publisher: PublisherConfig, + pub blossom: BlossomConfig, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -88,6 +89,13 @@ pub struct QBittorrentConfig { pub categories: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BlossomConfig { + pub enabled: bool, + /// Trusted Blossom server URL (no trailing slash), e.g. "https://blossom.example.com" + pub server: String, +} + impl Default for Config { fn default() -> Self { Config { @@ -134,6 +142,10 @@ impl Default for Config { method: "dht".into(), refresh_interval: "30m".into(), }, + blossom: BlossomConfig { + enabled: false, + server: String::new(), + }, publisher: PublisherConfig { enabled: false, outbox: vec![], diff --git a/src/db/migrations/007_blossom.sql b/src/db/migrations/007_blossom.sql new file mode 100644 index 0000000..83d41f0 --- /dev/null +++ b/src/db/migrations/007_blossom.sql @@ -0,0 +1,5 @@ +-- Blossom binary bridge support. +-- blossom_url: the URL of the .torrent file on the trusted Blossom server (from `url` tag in NIP-35 events, or after upload). +-- torrent_blob: raw .torrent file bytes fetched from that URL. +ALTER TABLE torrents ADD COLUMN blossom_url TEXT; +ALTER TABLE torrents ADD COLUMN torrent_blob BLOB; diff --git a/src/db/queries.rs b/src/db/queries.rs index 499a0cd..3ec5034 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -39,6 +39,7 @@ pub struct TorrentRecord { pub quality: String, pub source: String, pub raw_event: String, + pub blossom_url: String, pub trackers: Vec, pub tags: Vec, pub files: Vec, @@ -57,8 +58,8 @@ pub async fn insert_torrent(pool: &SqlitePool, r: &TorrentRecord) -> anyhow::Res "INSERT OR IGNORE INTO torrents (event_id, info_hash, pubkey, created_at, ingested_at, title, description, size_bytes, category, newznab_cat, imdb_id, tmdb_id, tvdb_id, - season, episode, quality, source, raw_event) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + season, episode, quality, source, raw_event, blossom_url) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ) .bind(&r.event_id) .bind(&r.info_hash) @@ -78,6 +79,7 @@ pub async fn insert_torrent(pool: &SqlitePool, r: &TorrentRecord) -> anyhow::Res .bind(null_str(&r.quality)) .bind(null_str(&r.source)) .bind(&r.raw_event) + .bind(null_str(&r.blossom_url)) .execute(&mut *tx) .await?; @@ -130,6 +132,35 @@ pub async fn insert_torrent(pool: &SqlitePool, r: &TorrentRecord) -> anyhow::Res Ok(()) } +/// Store a fetched .torrent blob. Also updates blossom_url if not already set. +pub async fn store_torrent_blob( + pool: &SqlitePool, + info_hash: &str, + blob: &[u8], + blossom_url: &str, +) -> anyhow::Result<()> { + sqlx::query( + "UPDATE torrents SET torrent_blob = ?, blossom_url = COALESCE(blossom_url, ?) + WHERE info_hash = ?", + ) + .bind(blob) + .bind(blossom_url) + .bind(info_hash) + .execute(pool) + .await?; + Ok(()) +} + +/// Retrieve the stored .torrent blob for a given info_hash, if any. +pub async fn get_torrent_blob(pool: &SqlitePool, info_hash: &str) -> anyhow::Result>> { + let row: Option<(Vec,)> = + sqlx::query_as("SELECT torrent_blob FROM torrents WHERE info_hash = ? AND torrent_blob IS NOT NULL LIMIT 1") + .bind(info_hash) + .fetch_optional(pool) + .await?; + Ok(row.map(|(b,)| b)) +} + // ─── Search ─────────────────────────────────────────────────────────────────── #[derive(Clone, Debug)] diff --git a/src/lib.rs b/src/lib.rs index 7b5c095..0562b07 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ +pub mod blossom; pub mod config; pub mod db; pub mod enrich; diff --git a/src/main.rs b/src/main.rs index 38a9cc3..9b36fdd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use anyhow::Context; use axum::{extract::State, routing::get, Json, Router}; use clap::Parser; -use kindexr::{config, db, enrich::tmdb::Enricher, nostr, publisher, torznab, ui, wot, AppState}; +use kindexr::{blossom::BlossomClient, config, db, enrich::tmdb::Enricher, nostr, publisher, torznab, ui, wot, AppState}; use serde_json::{json, Value}; use std::{ sync::{atomic::AtomicI32, Arc}, @@ -59,7 +59,15 @@ async fn main() -> anyhow::Result<()> { }; let enricher = Arc::new(Enricher::new(cfg.tmdb.api_key.clone())); - let reader = nostr::reader::Reader::new(cfg.clone(), pool.clone(), relay_count, enricher.clone()); + + let blossom: Option> = if cfg.blossom.enabled && !cfg.blossom.server.is_empty() { + info!(server = %cfg.blossom.server, "blossom client enabled"); + Some(Arc::new(BlossomClient::new(&cfg.blossom.server))) + } else { + None + }; + + let reader = nostr::reader::Reader::new(cfg.clone(), pool.clone(), relay_count, enricher.clone(), blossom.clone()); reader.start(); wot::follows::WotBuilder::new(cfg.clone(), pool.clone()).start(); @@ -73,6 +81,7 @@ async fn main() -> anyhow::Result<()> { pool.clone(), Arc::new(signer), enricher, + blossom.clone(), ); watcher.start(); info!("publisher watcher started"); diff --git a/src/nostr/parser.rs b/src/nostr/parser.rs index 71022ac..05ee274 100644 --- a/src/nostr/parser.rs +++ b/src/nostr/parser.rs @@ -44,6 +44,7 @@ pub fn parse(event: &Event) -> anyhow::Result { quality: String::new(), source: String::new(), raw_event: event.as_json(), + blossom_url: String::new(), trackers: vec![], tags: vec![], files: vec![], @@ -89,6 +90,12 @@ pub fn parse(event: &Event) -> anyhow::Result { "t" => { rec.tags.push(tag_vec[1].clone()); } + // `url` tag: Blossom URL pointing to the .torrent file (BUD-01) + "url" => { + if rec.blossom_url.is_empty() { + rec.blossom_url = tag_vec[1].clone(); + } + } _ => {} } } diff --git a/src/nostr/reader.rs b/src/nostr/reader.rs index 193ed0a..38f4e7f 100644 --- a/src/nostr/reader.rs +++ b/src/nostr/reader.rs @@ -7,18 +7,25 @@ use nostr_sdk::{prelude::*, RelayPoolNotification}; use sqlx::SqlitePool; use tracing::{debug, error, info, warn}; -use crate::{config::Config, db, enrich::tmdb::Enricher}; +use crate::{blossom::BlossomClient, config::Config, db, enrich::tmdb::Enricher}; pub struct Reader { cfg: Arc, pool: SqlitePool, connected: Arc, enricher: Arc, + blossom: Option>, } impl Reader { - pub fn new(cfg: Arc, pool: SqlitePool, connected: Arc, enricher: Arc) -> Self { - Reader { cfg, pool, connected, enricher } + pub fn new( + cfg: Arc, + pool: SqlitePool, + connected: Arc, + enricher: Arc, + blossom: Option>, + ) -> Self { + Reader { cfg, pool, connected, enricher, blossom } } pub fn start(self) { @@ -26,13 +33,14 @@ impl Reader { let pool = self.pool.clone(); let connected = self.connected.clone(); let enricher = self.enricher.clone(); + let blossom = self.blossom.clone(); tokio::spawn(async move { - run_reader(cfg, pool, connected, enricher).await; + run_reader(cfg, pool, connected, enricher, blossom).await; }); } } -async fn run_reader(cfg: Arc, pool: SqlitePool, connected: Arc, enricher: Arc) { +async fn run_reader(cfg: Arc, pool: SqlitePool, connected: Arc, enricher: Arc, blossom: Option>) { if cfg.relays.is_empty() { warn!("no relays configured; reader idle"); return; @@ -90,11 +98,12 @@ async fn run_reader(cfg: Arc, pool: SqlitePool, connected: Arc { match event.kind.as_u16() { - 2003 => handle_torrent(&pool, &cfg, &enricher, &event).await, + 2003 => handle_torrent(&pool, &cfg, &enricher, blossom.as_deref(), &event).await, 30004 => handle_curation_set(&pool, &event).await, 1984 => handle_report(&pool, &cfg, &event).await, _ => {} @@ -119,7 +128,7 @@ async fn run_reader(cfg: Arc, pool: SqlitePool, connected: Arc, event: &nostr_sdk::Event) { match super::parser::parse(event) { Ok(rec) => { // WoT / block / mute check @@ -170,6 +179,30 @@ async fn handle_torrent(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, ev enricher.enrich(&pool, &event_id, &parsed.clean, parsed.year, newznab_cat).await; }); } + + // Blossom: if the event included a `url` tag pointing to the .torrent file + // and Blossom is enabled, fetch and cache it for /torrent/.torrent + if blossom.is_some() && !rec.blossom_url.is_empty() { + let pool = pool.clone(); + let info_hash = rec.info_hash.clone(); + let url = rec.blossom_url.clone(); + tokio::spawn(async move { + let client = reqwest::Client::new(); + match client.get(&url).send().await { + Ok(resp) if resp.status().is_success() => { + match resp.bytes().await { + Ok(bytes) => { + let _ = db::store_torrent_blob(&pool, &info_hash, &bytes, &url).await; + debug!(info_hash = %info_hash, "blossom blob cached"); + } + Err(e) => debug!(url = %url, "blossom fetch body error: {e}"), + } + } + Ok(resp) => debug!(url = %url, "blossom fetch HTTP {}", resp.status()), + Err(e) => debug!(url = %url, "blossom fetch failed: {e}"), + } + }); + } } } } diff --git a/src/nostr/writer.rs b/src/nostr/writer.rs index b640947..90f8f26 100644 --- a/src/nostr/writer.rs +++ b/src/nostr/writer.rs @@ -57,6 +57,11 @@ pub fn build_event(rec: &TorrentRecord) -> anyhow::Result { tags.push(Tag::parse(["t", tag.as_str()])?); } + // BUD-01: include Blossom URL when available so other indexers can fetch the .torrent + if !rec.blossom_url.is_empty() { + tags.push(Tag::parse(["url", rec.blossom_url.as_str()])?); + } + Ok(EventBuilder::new(Kind::Custom(2003), rec.description.clone()).tags(tags)) } @@ -90,6 +95,7 @@ pub fn build_event_minimal( quality: String::new(), source: String::new(), raw_event: String::new(), + blossom_url: String::new(), trackers: trackers.to_vec(), tags: vec![], files: files.to_vec(), diff --git a/src/publisher/watcher.rs b/src/publisher/watcher.rs index 50b04c8..2dd6967 100644 --- a/src/publisher/watcher.rs +++ b/src/publisher/watcher.rs @@ -4,6 +4,7 @@ use sqlx::SqlitePool; use tracing::{debug, error, info, warn}; use crate::{ + blossom::BlossomClient, config::Config, db::{self, FileRecord, TorrentRecord}, enrich::{parser as title_parser, tmdb::Enricher}, @@ -17,11 +18,18 @@ pub struct Watcher { pool: SqlitePool, signer: Arc, enricher: Arc, + blossom: Option>, } impl Watcher { - pub fn new(cfg: Arc, pool: SqlitePool, signer: Arc, enricher: Arc) -> Self { - Watcher { cfg, pool, signer, enricher } + pub fn new( + cfg: Arc, + pool: SqlitePool, + signer: Arc, + enricher: Arc, + blossom: Option>, + ) -> Self { + Watcher { cfg, pool, signer, enricher, blossom } } pub fn start(self) { @@ -125,6 +133,7 @@ impl Watcher { quality: parsed.quality.unwrap_or_default(), source: parsed.source.unwrap_or_default(), raw_event: String::new(), + blossom_url: String::new(), trackers: vec![], tags: vec![], files: vec![], @@ -146,6 +155,25 @@ impl Watcher { } } + // Blossom upload: if we have the .torrent file and a Blossom server is configured, + // upload it and attach the URL to the record so it appears in the NIP-35 event. + if let (Some(blossom), Some(ref path)) = (&self.blossom, &item.torrent_path) { + match tokio::fs::read(path).await { + Ok(blob) => { + match blossom.upload(&blob, &self.signer).await { + Ok(url) => { + info!(info_hash = %rec.info_hash, url = %url, "blob uploaded to blossom"); + rec.blossom_url = url.clone(); + // Store blob locally so we can serve it via /torrent/.torrent + let _ = db::store_torrent_blob(&self.pool, &rec.info_hash, &blob, &url).await; + } + Err(e) => warn!(info_hash = %rec.info_hash, "blossom upload failed: {e}"), + } + } + Err(e) => warn!(path = %path, "failed to read torrent file for blossom upload: {e}"), + } + } + // Determine outbox relays for this category. let relays = outbox_for_category(&self.cfg, &item.category); if relays.is_empty() { @@ -221,6 +249,7 @@ pub fn build_from_torrent_file(path: &str, category: &str) -> anyhow::Result Router { .route("/ui/publishers/{pubkey}/unblock", post(publishers::unblock_handler)) .route("/ui/publishers/{pubkey}/mute", post(publishers::mute_handler)) .route("/ui/published", get(published::handler)) + .route("/torrent/{info_hash}.torrent", get(torrent_blob_handler)) +} + +/// Serve a cached .torrent file by info_hash. +async fn torrent_blob_handler( + State(state): State, + Path(info_hash): Path, +) -> Response { + match db::get_torrent_blob(&state.pool, &info_hash).await { + Ok(Some(blob)) => ( + StatusCode::OK, + [ + (header::CONTENT_TYPE, "application/x-bittorrent"), + ( + header::CONTENT_DISPOSITION, + &format!("attachment; filename=\"{info_hash}.torrent\""), + ), + ], + Bytes::from(blob), + ) + .into_response(), + Ok(None) => (StatusCode::NOT_FOUND, "torrent file not cached").into_response(), + Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "db error").into_response(), + } } // ─── Shared helpers ───────────────────────────────────────────────────────────