Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
13 kB · 376 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377//! registry projection over meta.git: repo list and the pending-genesis journal.
use std::sync::Arc;
use anyhow::Context as _;use base64::Engine as _;use knot_types::{KnotHostname, KnotServiceUrl, OwnerDid, RefName, RepoDid, RepoRkey};use serde::{Deserialize, Serialize};
use crate::config::ValidatedConfig;
pub const PENDING_REF: &str = "refs/atproto/plc/pending";
#[derive(Debug, Clone, Serialize, Deserialize)]pub struct PendingGenesis { pub did: RepoDid, pub operation: serde_json::Value, #[serde(default)] pub owner: Option<OwnerDid>, #[serde(default)] pub rkey: Option<RepoRkey>, #[serde(default)] pub name: Option<String>, #[serde(default)] pub source: Option<String>, #[serde(default)] pub seed: u64, #[serde(default)] pub created_at: i64,}
pub fn format_rfc3339(secs: i64) -> String { let days = secs.div_euclid(86400); let rem_secs = secs.rem_euclid(86400); let hour = rem_secs / 3600; let min = (rem_secs % 3600) / 60; let sec = rem_secs % 60;
let z = days + 719468; let era = (if z >= 0 { z } else { z - 146096 }) / 146097; let doe = (z - era * 146097) as u32; let yoe = (doe - doe / 1024 + doe / 1461 - doe / 146096) / 365; let y = (yoe as i64) + era * 400; let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); let mp = (5 * doy + 2) / 153; let d = doy - (153 * mp + 2) / 5 + 1; let m = if mp < 10 { mp + 3 } else { mp - 9 }; let y = if m <= 2 { y + 1 } else { y }; format!("{y:04}-{m:02}-{d:02}T{hour:02}:{min:02}:{sec:02}Z")}
pub fn load_pending_genesis(repo: &knot_git::Repo) -> anyhow::Result<Vec<PendingGenesis>> { let name = RefName::new(PENDING_REF)?; let Some(oid) = repo.find_ref(&name)? else { return Ok(Vec::new()); }; let object = repo.git().find_object(oid.object_id())?; Ok(serde_json::from_slice(&object.detach().data)?)}
pub fn save_pending_genesis( repo: &knot_git::Repo, entries: &[PendingGenesis],) -> anyhow::Result<()> { let name = RefName::new(PENDING_REF)?; let old = repo.find_ref(&name)?; if entries.is_empty() { if let Some(old) = old { repo.update_ref(&knot_git::RefUpdate::Delete { name, old })?; } return Ok(()); } let bytes = serde_json::to_vec(entries)?; let blob = repo.git().write_blob(&bytes)?; let new = knot_types::Oid::from(blob.detach()); let update = match old { Some(old) => knot_git::RefUpdate::Update { name, old, new }, None => knot_git::RefUpdate::Create { name, new }, }; repo.update_ref(&update)?; Ok(())}
pub fn enqueue_pending_genesis(repo: &knot_git::Repo, entry: PendingGenesis) -> anyhow::Result<()> { let mut entries = load_pending_genesis(repo)?; entries.retain(|existing| existing.did != entry.did); entries.push(entry); save_pending_genesis(repo, &entries)}
pub fn drop_pending_genesis(repo: &knot_git::Repo, did: &RepoDid) -> anyhow::Result<bool> { let mut entries = load_pending_genesis(repo)?; let before = entries.len(); entries.retain(|existing| existing.did != *did); let removed = entries.len() != before; if removed { save_pending_genesis(repo, &entries)?; } Ok(removed)}
pub async fn verify_did_document( transport: &knot_runtime::ReqwestHttp, plc_directory: &url::Url, did: &RepoDid, hostname: &KnotHostname,) -> anyhow::Result<()> { let base = plc_directory.as_str().trim_end_matches('/'); let url = url::Url::parse(&format!("{base}/{}", did.as_str())) .context("construct plc document url")?;
if url.scheme() != "https" { anyhow::bail!("insecure scheme in plc directory url: {url}"); } let blocked_ip = match url.host() { Some(url::Host::Ipv4(ip)) => knot_runtime::is_blocked_ip(ip.into()).then(|| ip.to_string()), Some(url::Host::Ipv6(ip)) => knot_runtime::is_blocked_ip(ip.into()).then(|| ip.to_string()), _ => None, }; if let Some(host) = blocked_ip { anyhow::bail!("blocked host in plc directory url: {host}"); }
let request = knot_runtime::HttpRequest::get(url); let response = knot_runtime::HttpTransport::execute(transport, request).await?; if !response.status.is_success() { anyhow::bail!("failed to fetch DID document: status {}", response.status); } let doc: serde_json::Value = serde_json::from_slice(response.body.as_ref()).context("parse DID document JSON")?;
let services = doc .get("service") .and_then(|s| s.as_array()) .ok_or_else(|| anyhow::anyhow!("DID document has no service list"))?;
let found = services.iter().any(|service| { service.get("type").and_then(|t| t.as_str()) == Some("TangledKnot") && service .get("serviceEndpoint") .and_then(|e| e.as_str()) .and_then(|endpoint| url::Url::parse(endpoint).ok()) .is_some_and(|endpoint_url| { endpoint_url.scheme() == "https" && endpoint_url.host_str() == Some(hostname.as_str()) }) });
anyhow::ensure!( found, "DID document for {did} does not declare a TangledKnot service pointing to {hostname}" ); Ok(())}
pub async fn list(config: Arc<ValidatedConfig>) -> anyhow::Result<()> { let layout = config.layout()?; let meta_path = layout.meta_path(&config.knot_did)?; let index = knot_index::Index::new(&meta_path, layout.clone()); index.rebuild().context("rebuild index")?;
let meta_repo = layout .bootstrap_meta(&config.knot_did) .context("open meta.git")?; let cob_store = knot_cob::CobStore::new(&meta_repo); let cob_objects = cob_store.list::<knot_cobs::RepoRegistryCob>()?; let registry_state = cob_objects .first() .map(|&object| cob_store.get::<knot_cobs::RepoRegistryCob>(object)) .transpose()? .map(|cob| cob.into_state());
let hosted = index.hosted_repos(); for did in hosted { let knot_index::Resolved::Ready(Some(ownership)) = index.ownership_of(&did) else { continue; }; let name = registry_state .as_ref() .and_then(|reg| reg.record_of(&did)) .map(|rec| rec.name.to_string()) .unwrap_or_else(|| ownership.rkey.to_string());
let head_str = layout .open(&did) .ok() .and_then(|repo| repo.head()) .map(|head| head.target.to_string()) .unwrap_or_else(|| "-".to_string());
let source_path = layout.guarded_path(&did)?; println!( "{did} {} {} {name} {head_str} {}", ownership.owner, ownership.rkey, source_path.display() ); }
Ok(())}
pub async fn plc_sync(config: Arc<ValidatedConfig>) -> anyhow::Result<()> { let layout = config.layout()?; let meta_path = layout.meta_path(&config.knot_did)?; let meta_repo = layout .bootstrap_meta(&config.knot_did) .context("open meta.git")?;
let entries = load_pending_genesis(&meta_repo)?; if entries.is_empty() { return Ok(()); }
let master_key = knot_secrets::MasterKey::new( base64::engine::general_purpose::STANDARD .decode( std::env::var(&config.master_key_env) .with_context(|| format!("read master key from {}", config.master_key_env))? .trim(), ) .context("decode master key as base64")?, ) .context("load master key")?;
let secrets = knot_secrets::SealedStore::open( &config.sealed_key_file, &master_key, Box::new(knot_runtime::OsEntropy), ) .context("open sealed key store")?;
secrets .ensure(&config.knot_did) .context("ensure knot signing key")?;
let knot_signer = secrets .signer(&config.knot_did) .context("get knot signer")?;
let service_url = KnotServiceUrl::new(format!("https://{}", config.hostname)).context("knot service url")?;
let mut failed = 0; let mut verified = Vec::new();
for entry in &entries { let (Some(owner), Some(rkey)) = (&entry.owner, &entry.rkey) else { failed += 1; eprintln!( "pending genesis for {}: missing owner/rkey to reconstruct operation", entry.did ); continue; };
let entropy = knot_runtime::SeededEntropy::new(entry.seed); let nonce = knot_atproto::MintNonce::mint(&entropy, owner, rkey); let prepared = match knot_atproto::prepare_repo_did(&knot_signer, &service_url, &nonce) { Ok(prep) => prep, Err(e) => { failed += 1; eprintln!( "pending genesis for {}: prepare_repo_did failed: {e:#}", entry.did ); continue; } };
let transport = match knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default()) { Ok(t) => t, Err(e) => { failed += 1; eprintln!("init http transport failed: {e:#}"); continue; } };
let atproto = match knot_atproto::PlcDirectory::new(config.plc_directory.clone()) { Ok(dir) => knot_atproto::Atproto::new( transport, knot_runtime::SystemClock, config.knot_did.clone(), dir, ), Err(e) => { failed += 1; eprintln!("invalid plc directory {}: {e:#}", config.plc_directory); continue; } };
let submit_res = atproto .submit_plc_operation(&prepared.did.clone(), prepared.operation()) .await;
let verify_transport = match knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default()) { Ok(t) => t, Err(e) => { failed += 1; eprintln!("init http transport for verify failed: {e:#}"); continue; } };
let verify_res = verify_did_document( &verify_transport, &config.plc_directory, &entry.did, &config.hostname, ) .await;
match verify_res { Ok(()) => { verified.push(entry.did.clone()); } Err(verify_err) => { failed += 1; match submit_res { Err(submit_err) => eprintln!( "pending genesis for {}: submission failed: {submit_err:#}; verification failed: {verify_err:#}", entry.did ), Ok(_) => eprintln!( "pending genesis for {}: verification failed: {verify_err:#}", entry.did ), } } } }
for did in &verified { drop_pending_genesis(&meta_repo, did)?; }
if !verified.is_empty() { let index = knot_index::Index::new(&meta_path, layout); let _ = index.rebuild(); }
anyhow::ensure!( failed == 0, "{failed} pending genesis operation(s) failed to sync" );
Ok(())}
#[cfg(test)]mod live_plc_tests { use super::*;
/// read-only acceptance probe against the REAL plc.directory (egress-gated: set /// KNOT_STATIC_LIVE_PLC=1). verifies our DID-document predicate accepts a production /// Tangled repo DID whose TangledKnot service matches its hosting knot — the exact /// check `plc sync` runs after genesis submission. #[tokio::test] async fn live_plc_repo_did_has_tangled_knot_service_on_host() { if std::env::var("KNOT_STATIC_LIVE_PLC").unwrap_or_default() != "1" { return; } let directory = url::Url::parse("https://plc.directory").unwrap(); let transport = knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default()) .expect("http transport"); // Real production repo did:plc minted by the production knot1.tangled.sh. let did = RepoDid::new("did:plc:yovatos6i5a4fonfcxusiqqw".to_string()).unwrap(); let hostname = knot_types::KnotHostname::new("knot1.tangled.sh".to_string()).unwrap(); verify_did_document(&transport, &directory, &did, &hostname) .await .expect("live plc resolution must carry a TangledKnot service on the host"); }}