use std::collections::{BTreeSet, HashMap, VecDeque};
use std::io::{self, BufRead, Write};
use std::path::PathBuf;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use rns_core::msgpack::{self, Value};
use rns_core::types::{DestHash, LinkId};
use rns_net::{AnnouncedIdentity, Callbacks, PacketHash, RnsNode};
use crate::config::ClientConfig;
use crate::logging;
use crate::protocol::{self, RefUpdate};
use crate::util::{
default_reticulum_dir, default_rngit_dir, load_or_create_identity, parse_rns_url_with_aliases,
};
use crate::{git, Error, Result};
const LINK_IDENTIFY_SETTLE_DELAY: Duration = Duration::from_millis(750);
const NOT_IDENTIFIED_RETRY_DELAY: Duration = Duration::from_millis(750);
const NOT_IDENTIFIED_RETRIES: usize = 5;
pub fn main<I>(args: I) -> Result<()>
where
I: IntoIterator<Item = String>,
{
let options = ClientOptions::parse(args)?;
let rngit_dir = options.config_dir.unwrap_or_else(default_rngit_dir);
let rns_dir = options.rns_config_dir.or_else(default_reticulum_dir);
let config = ClientConfig::load_or_create_for_run(rngit_dir, rns_dir)?;
logging::init_file_logger(&config.dir.join("client_log"), config.log_level)?;
let (dest_hash, repository) =
parse_rns_url_with_aliases(&options.url, &config.destination_aliases)?;
let helper = RemoteHelper::connect(config, dest_hash)?;
helper.run(repository)
}
struct RemoteHelper {
client: SyncClient,
remote_refs: Mutex<HashMap<String, String>>,
}
impl RemoteHelper {
fn connect(config: ClientConfig, dest_hash: [u8; 16]) -> Result<Self> {
Ok(Self {
client: SyncClient::connect(config, dest_hash)?,
remote_refs: Mutex::new(HashMap::new()),
})
}
fn run(self, repository: String) -> Result<()> {
let stdin = io::stdin();
let stdout = io::stdout();
self.run_io(repository, stdin.lock(), stdout.lock())
}
fn run_io<R: BufRead, W: Write>(
&self,
repository: String,
mut input: R,
mut output: W,
) -> Result<()> {
let mut line = String::new();
let mut fetch_refs = Vec::new();
let mut push_specs = Vec::new();
loop {
line.clear();
if input.read_line(&mut line)? == 0 {
break;
}
let command = line.trim_end();
if command.is_empty() {
let completed_batch = !fetch_refs.is_empty() || !push_specs.is_empty();
if !fetch_refs.is_empty() {
self.fetch(&repository, &fetch_refs)?;
fetch_refs.clear();
writeln!(output)?;
output.flush()?;
}
if !push_specs.is_empty() {
for spec in push_specs.drain(..) {
self.push(&repository, &spec)?;
writeln!(output, "ok {}", spec.remote)?;
}
writeln!(output)?;
output.flush()?;
}
if completed_batch {
break;
}
continue;
}
if command == "capabilities" {
writeln!(output, "option")?;
writeln!(output, "list")?;
writeln!(output, "fetch")?;
writeln!(output, "push")?;
writeln!(output)?;
} else if command == "list" || command == "list for-push" {
let refs = self.list(&repository)?;
output.write_all(&refs)?;
writeln!(output)?;
} else if let Some(rest) = command.strip_prefix("option ") {
let _ = rest;
writeln!(output, "ok")?;
} else if let Some(rest) = command.strip_prefix("fetch ") {
fetch_refs.push(parse_fetch_command(rest)?);
} else if let Some(rest) = command.strip_prefix("push ") {
push_specs.push(parse_push_spec(rest)?);
} else {
writeln!(output, "error unsupported command")?;
}
output.flush()?;
}
Ok(())
}
fn list(&self, repository: &str) -> Result<Vec<u8>> {
let response = self.request_with_ident_retry(
protocol::PATH_LIST,
protocol::repository_request(repository),
)?;
let bytes = protocol::response_bin(&response.data)?;
let refs = decode_status(bytes)?;
self.record_remote_refs(&refs);
Ok(refs)
}
fn fetch(&self, repository: &str, refs: &[FetchRef]) -> Result<()> {
let have = self.fetch_have_set(refs);
let requested_refs = refs
.iter()
.map(|r| (r.sha.clone(), r.name.clone()))
.collect::<Vec<_>>();
let response = self.request_with_ident_retry(
protocol::PATH_FETCH,
protocol::fetch_request_for_refs(repository, &have, &requested_refs),
)?;
if let Some(metadata) = response.metadata {
ensure_metadata_ok(&metadata)?;
let bundle = protocol::response_bin(&response.data)?;
git::fetch_bundle_into_local(
&bundle,
&refs.iter().map(|r| r.name.clone()).collect::<Vec<_>>(),
)?;
return Ok(());
}
let bytes = protocol::response_bin(&response.data)?;
let bundle = decode_status(bytes)?;
if !bundle.is_empty() {
git::fetch_bundle_into_local(
&bundle,
&refs.iter().map(|r| r.name.clone()).collect::<Vec<_>>(),
)?;
}
Ok(())
}
fn request_with_ident_retry(&self, path: &str, data: Vec<u8>) -> Result<Response> {
for attempt in 0..=NOT_IDENTIFIED_RETRIES {
let response = self.client.request(path, data.clone())?;
if response.metadata.is_none() {
if let Ok(bytes) = protocol::response_bin(&response.data) {
if is_not_identified_status(&bytes) && attempt < NOT_IDENTIFIED_RETRIES {
self.client.identify()?;
std::thread::sleep(NOT_IDENTIFIED_RETRY_DELAY);
continue;
}
}
}
return Ok(response);
}
unreachable!("not-identified retry loop always returns")
}
fn push(&self, repository: &str, spec: &PushSpec) -> Result<()> {
let mut updates = Vec::new();
let mut bundle_refs = Vec::new();
if spec.local.is_empty() {
updates.push(RefUpdate {
refname: spec.remote.clone(),
old: None,
new: None,
force: spec.force,
});
} else {
let sha = git::local_ref_sha(&spec.local)?
.ok_or_else(|| Error::msg(format!("unknown local ref {}", spec.local)))?;
bundle_refs.push(spec.local.clone());
updates.push(RefUpdate {
refname: spec.remote.clone(),
old: None,
new: Some(sha),
force: spec.force,
});
}
let exclusions = self.push_exclusion_set();
let bundle = git::create_local_bundle(&bundle_refs, &exclusions)?;
let response = self.client.request(
protocol::PATH_PUSH,
protocol::push_request(repository, bundle, updates),
)?;
let bytes = protocol::response_bin(&response.data)?;
let _ = decode_status(bytes)?;
Ok(())
}
fn record_remote_refs(&self, refs: &[u8]) {
*self.remote_refs.lock().unwrap() = parse_remote_refs(refs);
}
fn fetch_have_set(&self, refs: &[FetchRef]) -> Vec<String> {
let remote_refs = self.remote_refs.lock().unwrap();
build_fetch_have_set(
&remote_refs,
refs,
|refname| git::local_ref_sha(refname).ok().flatten(),
git::object_exists_local,
)
}
fn push_exclusion_set(&self) -> Vec<String> {
let remote_refs = self.remote_refs.lock().unwrap();
build_push_exclusion_set(&remote_refs, git::object_exists_local)
}
}
fn parse_remote_refs(refs: &[u8]) -> HashMap<String, String> {
let Ok(text) = std::str::from_utf8(refs) else {
return HashMap::new();
};
text.lines()
.filter_map(|line| {
let (sha, name) = line.split_once(' ')?;
(name != "HEAD").then(|| (name.to_string(), sha.to_string()))
})
.collect()
}
fn build_fetch_have_set(
remote_refs: &HashMap<String, String>,
refs: &[FetchRef],
resolve_local_ref: impl Fn(&str) -> Option<String>,
object_exists: impl Fn(&str) -> bool,
) -> Vec<String> {
let mut have = BTreeSet::new();
for fetch_ref in refs {
if let Some(local_sha) = resolve_local_ref(&fetch_ref.name) {
if local_sha != fetch_ref.sha {
have.insert(local_sha);
}
}
}
for (refname, remote_sha) in remote_refs {
if object_exists(remote_sha) {
have.insert(remote_sha.clone());
continue;
}
if let Some(local_sha) = resolve_local_ref(refname) {
if &local_sha == remote_sha {
have.insert(local_sha);
}
}
}
have.into_iter().collect()
}
fn build_push_exclusion_set(
remote_refs: &HashMap<String, String>,
object_exists: impl Fn(&str) -> bool,
) -> Vec<String> {
remote_refs
.values()
.filter(|sha| object_exists(sha))
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.collect()
}
pub(crate) struct SyncClient {
node: RnsNode,
link_id: [u8; 16],
identity_private_key: [u8; 64],
state: Arc<(Mutex<ClientState>, Condvar)>,
request_timeout: Duration,
}
impl SyncClient {
pub(crate) fn connect(config: ClientConfig, dest_hash: [u8; 16]) -> Result<Self> {
let callbacks = SharedCallbacks::default();
let state = callbacks.state.clone();
let node = RnsNode::from_config(config.reticulum_dir.as_deref(), Box::new(callbacks))?;
let client_identity = load_or_create_identity(&config.identity_path)?;
let dest = DestHash(dest_hash);
eprintln!("Requesting path to {}...", crate::util::hex(&dest_hash));
node.request_path(&dest)
.map_err(|_| Error::msg("failed to request destination path"))?;
let configured_timeout = Duration::from_secs(config.connect_timeout_secs);
let medium_timeout =
Duration::from_secs_f64(node.medium_path_timeout().unwrap_or(0.0).max(0.0));
let deadline = Instant::now() + configured_timeout.max(medium_timeout);
while !node.has_path(&dest).unwrap_or(false) && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(250));
}
if !node.has_path(&dest).unwrap_or(false) {
return Err(Error::msg("destination path resolution timed out"));
}
eprintln!("Path resolved.");
let recalled = node
.recall_identity(&dest)
.map_err(|_| Error::msg("failed to recall destination identity"))?
.ok_or_else(|| Error::msg("destination identity is unknown"))?;
let sig_pub: [u8; 32] = recalled.public_key[32..64].try_into().unwrap();
let link_id = node
.create_link(dest_hash, sig_pub)
.map_err(|_| Error::msg("failed to create RNS link"))?;
eprintln!("Establishing link...");
let hops = node.hops_to(&dest).unwrap_or(None).unwrap_or(1);
let link_timeout =
Duration::from_secs_f64(rns_core::link::keepalive::compute_establishment_timeout(
rns_core::constants::LINK_ESTABLISHMENT_TIMEOUT_PER_HOP,
hops,
));
wait_for_link(&state, link_id, configured_timeout.max(link_timeout))?;
eprintln!("Link established.");
let private_key = client_identity
.get_private_key()
.ok_or_else(|| Error::msg("client identity has no private key"))?;
std::thread::sleep(LINK_IDENTIFY_SETTLE_DELAY);
identify_on_link(&node, link_id, private_key)?;
std::thread::sleep(LINK_IDENTIFY_SETTLE_DELAY);
Ok(Self {
node,
link_id,
identity_private_key: private_key,
state,
request_timeout: Duration::from_secs(config.request_timeout_secs),
})
}
pub(crate) fn identify(&self) -> Result<()> {
identify_on_link(&self.node, self.link_id, self.identity_private_key)
}
pub(crate) fn request(&self, path: &str, data: Vec<u8>) -> Result<Response> {
self.request_with_timeout(path, data, self.request_timeout)
}
pub(crate) fn request_with_timeout(
&self,
path: &str,
data: Vec<u8>,
timeout: Duration,
) -> Result<Response> {
{
let (lock, _) = &*self.state;
let mut state = lock.lock().unwrap();
state.responses.clear();
state.progress = ProgressState::default();
}
self.node
.send_request(self.link_id, path, &data)
.map_err(|_| Error::msg("failed to send request"))?;
let deadline = Instant::now() + timeout;
let (lock, cv) = &*self.state;
let mut state = lock.lock().unwrap();
loop {
if let Some(response) = state.responses.pop_front() {
return Ok(response);
}
let now = Instant::now();
if now >= deadline {
return Err(Error::msg("request timed out"));
}
let wait = deadline.saturating_duration_since(now);
let (next, _) = cv.wait_timeout(state, wait).unwrap();
state = next;
}
}
}
fn identify_on_link(node: &RnsNode, link_id: [u8; 16], private_key: [u8; 64]) -> Result<()> {
node.identify_on_link(link_id, private_key)
.map_err(|_| Error::msg("failed to identify on RNS link"))
}
#[derive(Default)]
struct SharedCallbacks {
state: Arc<(Mutex<ClientState>, Condvar)>,
}
#[derive(Default)]
struct ClientState {
established: Vec<[u8; 16]>,
responses: VecDeque<Response>,
progress: ProgressState,
}
#[derive(Default)]
struct ProgressState {
started_at: Option<Instant>,
last_percent: Option<usize>,
}
#[derive(Debug, Clone)]
pub(crate) struct Response {
pub(crate) data: Vec<u8>,
pub(crate) metadata: Option<Vec<u8>>,
}
impl Callbacks for SharedCallbacks {
fn on_announce(&mut self, _announced: AnnouncedIdentity) {}
fn on_path_updated(&mut self, _dest_hash: DestHash, _hops: u8) {}
fn on_local_delivery(&mut self, _dest_hash: DestHash, _raw: Vec<u8>, _packet_hash: PacketHash) {
}
fn on_link_established(
&mut self,
link_id: LinkId,
_dest_hash: DestHash,
_rtt: f64,
_is_initiator: bool,
) {
let (lock, cv) = &*self.state;
lock.lock().unwrap().established.push(link_id.0);
cv.notify_all();
}
fn on_response_with_metadata(
&mut self,
_link_id: LinkId,
_request_id: [u8; 16],
data: Vec<u8>,
metadata: Option<Vec<u8>>,
) {
let (lock, cv) = &*self.state;
lock.lock()
.unwrap()
.responses
.push_back(Response { data, metadata });
cv.notify_all();
}
fn on_resource_progress(&mut self, _link_id: LinkId, received: usize, total: usize) {
if total == 0 {
return;
}
let (lock, _) = &*self.state;
let mut state = lock.lock().unwrap();
let progress = &mut state.progress;
let started_at = *progress.started_at.get_or_insert_with(Instant::now);
let percent = ((received.saturating_mul(100)) / total).min(100);
if progress.last_percent == Some(percent) && received < total {
return;
}
progress.last_percent = Some(percent);
let elapsed = started_at.elapsed().as_secs_f64().max(0.001);
let rate = received as f64 / elapsed;
eprintln!("rns: transfer {percent}% ({received}/{total} parts, {rate:.1} parts/s)");
if received >= total {
progress.started_at = None;
progress.last_percent = None;
}
}
}
fn wait_for_link(
state: &Arc<(Mutex<ClientState>, Condvar)>,
link_id: [u8; 16],
timeout: Duration,
) -> Result<()> {
let deadline = Instant::now() + timeout;
let (lock, cv) = &**state;
let mut state = lock.lock().unwrap();
loop {
if state.established.contains(&link_id) {
return Ok(());
}
let now = Instant::now();
if now >= deadline {
return Err(Error::msg("link establishment timed out"));
}
let (next, _) = cv
.wait_timeout(state, deadline.saturating_duration_since(now))
.unwrap();
state = next;
}
}
fn is_not_identified_status(bytes: &[u8]) -> bool {
matches!(
bytes.split_first(),
Some((&protocol::RES_DISALLOWED, body)) if body == b"Not identified"
)
}
pub(crate) fn decode_status(bytes: Vec<u8>) -> Result<Vec<u8>> {
let Some((&code, body)) = bytes.split_first() else {
return Err(Error::msg("empty response"));
};
if code == protocol::RES_OK {
Ok(body.to_vec())
} else {
Err(Error::msg(format!(
"remote returned status 0x{code:02x}: {}",
String::from_utf8_lossy(body)
)))
}
}
pub(crate) fn ensure_metadata_ok(metadata: &[u8]) -> Result<()> {
let value = msgpack::unpack_exact(metadata)
.map_err(|e| Error::msg(format!("invalid response metadata: {e}")))?;
let Some(map) = value.as_map() else {
return Err(Error::msg("response metadata is not a map"));
};
let code = map.iter().find_map(|(key, value)| {
if matches!(key, Value::UInt(v) if *v == protocol::IDX_RESULT_CODE) {
value.as_uint().map(|v| v as u8)
} else {
None
}
});
match code {
Some(protocol::RES_OK) => Ok(()),
Some(code) => Err(Error::msg(format!("remote returned status 0x{code:02x}"))),
None => Err(Error::msg("response metadata missing status code")),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct FetchRef {
sha: String,
name: String,
}
fn parse_fetch_command(input: &str) -> Result<FetchRef> {
let mut parts = input.split_whitespace();
let sha = parts
.next()
.ok_or_else(|| Error::msg("fetch command missing sha"))?;
let name = parts
.next()
.ok_or_else(|| Error::msg("fetch command missing ref"))?;
Ok(FetchRef {
sha: sha.to_string(),
name: name.to_string(),
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct PushSpec {
local: String,
remote: String,
force: bool,
}
fn parse_push_spec(input: &str) -> Result<PushSpec> {
let (force, spec) = input
.strip_prefix('+')
.map(|s| (true, s))
.unwrap_or((false, input));
let (local, remote) = spec
.split_once(':')
.ok_or_else(|| Error::msg("push spec must be <local>:<remote>"))?;
Ok(PushSpec {
local: local.to_string(),
remote: remote.to_string(),
force,
})
}
#[derive(Debug, Default)]
struct ClientOptions {
config_dir: Option<PathBuf>,
rns_config_dir: Option<PathBuf>,
url: String,
}
impl ClientOptions {
fn parse<I>(args: I) -> Result<Self>
where
I: IntoIterator<Item = String>,
{
let mut options = ClientOptions::default();
let mut positional = Vec::new();
let mut args = args.into_iter();
while let Some(arg) = args.next() {
match arg.as_str() {
"-c" | "--config" => {
options.config_dir = Some(PathBuf::from(
args.next()
.ok_or_else(|| Error::msg("missing config path"))?,
));
}
"--rnsconfig" => {
options.rns_config_dir = Some(PathBuf::from(
args.next()
.ok_or_else(|| Error::msg("missing RNS config path"))?,
));
}
"-h" | "--help" => return Err(Error::msg(usage())),
other => positional.push(other.to_string()),
}
}
options.url = positional
.last()
.cloned()
.ok_or_else(|| Error::msg(usage()))?;
Ok(options)
}
}
fn usage() -> &'static str {
"usage: git-remote-rns [--config DIR] [--rnsconfig DIR] <remote-name> <rns://destination/repo>"
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::{BufRead, BufReader};
use std::process::{Child, Command, Stdio};
use std::sync::mpsc;
#[test]
fn parses_fetch_command() {
assert_eq!(
parse_fetch_command("abc refs/heads/main").unwrap(),
FetchRef {
sha: "abc".into(),
name: "refs/heads/main".into()
}
);
}
#[test]
fn parses_forced_push_spec() {
assert_eq!(
parse_push_spec("+refs/heads/main:refs/heads/main").unwrap(),
PushSpec {
local: "refs/heads/main".into(),
remote: "refs/heads/main".into(),
force: true
}
);
}
#[test]
fn decodes_status_payload() {
assert_eq!(
decode_status(protocol::status_bytes(protocol::RES_OK, b"refs")).unwrap(),
b"refs"
);
assert!(decode_status(protocol::status_bytes(protocol::RES_NOT_FOUND, b"no")).is_err());
}
#[test]
fn metadata_status_ok_is_accepted() {
assert!(ensure_metadata_ok(&protocol::metadata_status(protocol::RES_OK)).is_ok());
assert!(ensure_metadata_ok(&protocol::metadata_status(protocol::RES_REMOTE_FAIL)).is_err());
}
#[test]
fn parses_remote_refs_and_ignores_head() {
let refs = parse_remote_refs(
b"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa HEAD\nbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb refs/heads/main\nmalformed\n",
);
assert_eq!(refs.len(), 1);
assert_eq!(
refs["refs/heads/main"],
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
);
}
#[test]
fn fetch_have_set_uses_per_ref_and_global_matching_haves() {
let remote_refs = HashMap::from([
(
"refs/heads/main".to_string(),
"1111111111111111111111111111111111111111".to_string(),
),
(
"refs/tags/v1".to_string(),
"2222222222222222222222222222222222222222".to_string(),
),
]);
let fetch_refs = vec![FetchRef {
sha: "3333333333333333333333333333333333333333".into(),
name: "refs/heads/feature".into(),
}];
let have = build_fetch_have_set(
&remote_refs,
&fetch_refs,
|name| match name {
"refs/heads/main" => Some("1111111111111111111111111111111111111111".into()),
"refs/tags/v1" => Some("not-remote".into()),
"refs/heads/feature" => Some("4444444444444444444444444444444444444444".into()),
_ => None,
},
|sha| sha == "2222222222222222222222222222222222222222",
);
assert_eq!(
have,
vec![
"1111111111111111111111111111111111111111",
"2222222222222222222222222222222222222222",
"4444444444444444444444444444444444444444"
]
);
}
#[test]
fn fetch_have_set_includes_existing_remote_objects_without_matching_refs() {
let remote_refs = HashMap::from([(
"refs/tags/v1".to_string(),
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
)]);
let have = build_fetch_have_set(
&remote_refs,
&[],
|_| None,
|sha| sha == "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
);
assert_eq!(have, vec!["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]);
}
#[test]
fn push_exclusion_set_keeps_only_local_known_remote_objects() {
let remote_refs = HashMap::from([
("refs/heads/main".to_string(), "aaaa".to_string()),
("refs/heads/dev".to_string(), "bbbb".to_string()),
("refs/tags/v1".to_string(), "aaaa".to_string()),
]);
let exclusions = build_push_exclusion_set(&remote_refs, |sha| sha == "aaaa");
assert_eq!(exclusions, vec!["aaaa"]);
}
#[test]
fn sync_client_reidentifies_after_not_identified_response() {
let _guard = python_rns_test_guard();
if !python_rns_available() {
eprintln!("Skipping: Python RNS not available");
return;
}
let mut python = PythonRngitPeer::spawn_reidentify_required();
let ready = python.wait_for_ready();
let tmp = tempfile::tempdir().unwrap();
let rns_dir = tmp.path().join("client-rns");
std::fs::create_dir_all(&rns_dir).unwrap();
std::fs::write(rns_dir.join("config"), client_rns_config(ready.port)).unwrap();
let client = SyncClient::connect(
ClientConfig {
dir: tmp.path().join("rngit"),
reticulum_dir: Some(rns_dir),
identity_path: tmp.path().join("client_identity"),
connect_timeout_secs: 10,
request_timeout_secs: 10,
destination_aliases: Default::default(),
log_level: crate::logging::DEFAULT_LOG_LEVEL,
},
parse_hex_16(&ready.dest_hash),
)
.expect("Rust client should connect to Python Reticulum peer");
let helper = RemoteHelper {
client,
remote_refs: Mutex::new(HashMap::new()),
};
let response = helper
.request_with_ident_retry(
protocol::PATH_LIST,
protocol::repository_request("group/repo"),
)
.expect("Rust client should re-identify and retry the request");
let response = protocol::response_bin(&response.data).unwrap();
let refs = String::from_utf8(decode_status(response).unwrap()).unwrap();
assert_eq!(
refs,
"1111111111111111111111111111111111111111 refs/heads/main\n"
);
let first = python.wait_for_event("REQUEST");
assert!(
first.ends_with(" 1 1"),
"unexpected first request event: {first}"
);
let second = python.wait_for_event("REQUEST");
assert!(
second.ends_with(" 2 2"),
"retry happened without a second LINKIDENTIFY: {second}"
);
}
#[test]
fn sync_client_identifies_before_first_python_request() {
let _guard = python_rns_test_guard();
if !python_rns_available() {
eprintln!("Skipping: Python RNS not available");
return;
}
let mut python = PythonRngitPeer::spawn();
let ready = python.wait_for_ready();
let tmp = tempfile::tempdir().unwrap();
let rns_dir = tmp.path().join("client-rns");
std::fs::create_dir_all(&rns_dir).unwrap();
std::fs::write(rns_dir.join("config"), client_rns_config(ready.port)).unwrap();
let client = SyncClient::connect(
ClientConfig {
dir: tmp.path().join("rngit"),
reticulum_dir: Some(rns_dir),
identity_path: tmp.path().join("client_identity"),
connect_timeout_secs: 10,
request_timeout_secs: 10,
destination_aliases: Default::default(),
log_level: crate::logging::DEFAULT_LOG_LEVEL,
},
parse_hex_16(&ready.dest_hash),
)
.expect("Rust client should connect to Python Reticulum peer");
let response = client
.request(
protocol::PATH_LIST,
protocol::repository_request("group/repo"),
)
.expect("Python peer should answer the first request");
let response = protocol::response_bin(&response.data).unwrap();
let refs = String::from_utf8(decode_status(response).unwrap()).unwrap();
assert_eq!(
refs,
"1111111111111111111111111111111111111111 refs/heads/main\n"
);
let request = python.wait_for_event("REQUEST");
let remote = request
.split_whitespace()
.nth(2)
.expect("REQUEST event should include remote identity hash");
assert_ne!(remote, "none", "Python saw an unidentified first request");
}
struct PythonReady {
port: u16,
dest_hash: String,
}
struct PythonRngitPeer {
child: Child,
events: mpsc::Receiver<String>,
}
impl PythonRngitPeer {
fn spawn() -> Self {
Self::spawn_with_reidentify_required(false)
}
fn spawn_reidentify_required() -> Self {
Self::spawn_with_reidentify_required(true)
}
fn spawn_with_reidentify_required(reidentify_required: bool) -> Self {
let mut command = Command::new("python3");
command.args(["-c", PYTHON_RNGIT_PEER_SCRIPT]);
if reidentify_required {
command.env("RNGIT_TEST_REQUIRE_REIDENTIFY", "1");
}
let mut child = command
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("failed to start Python RNS process");
let stdout = child.stdout.take().unwrap();
let stderr = child.stderr.take().unwrap();
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
for line in BufReader::new(stdout).lines().map_while(|line| line.ok()) {
let _ = tx.send(line);
}
});
std::thread::spawn(move || {
for line in BufReader::new(stderr).lines().map_while(|line| line.ok()) {
eprintln!("python stderr: {line}");
}
});
Self { child, events: rx }
}
fn wait_for_ready(&mut self) -> PythonReady {
let event = self.wait_for_event("READY");
let mut parts = event.split_whitespace();
assert_eq!(parts.next(), Some("READY"));
let port = parts.next().unwrap().parse().unwrap();
let dest_hash = parts.next().unwrap().to_string();
PythonReady { port, dest_hash }
}
fn wait_for_event(&mut self, prefix: &str) -> String {
let deadline = Instant::now() + Duration::from_secs(10);
loop {
let remaining = deadline
.checked_duration_since(Instant::now())
.expect("timed out waiting for Python event");
let event = self
.events
.recv_timeout(remaining)
.expect("timed out waiting for Python event");
if event.starts_with(prefix) {
return event;
}
}
}
}
impl Drop for PythonRngitPeer {
fn drop(&mut self) {
if let Some(stdin) = self.child.stdin.as_mut() {
use std::io::Write;
let _ = writeln!(stdin, "stop");
let _ = stdin.flush();
}
let _ = self.child.kill();
let _ = self.child.wait();
}
}
static PYTHON_RNS_TEST_LOCK: std::sync::OnceLock<std::sync::Mutex<()>> =
std::sync::OnceLock::new();
fn python_rns_test_guard() -> std::sync::MutexGuard<'static, ()> {
PYTHON_RNS_TEST_LOCK
.get_or_init(|| std::sync::Mutex::new(()))
.lock()
.unwrap()
}
fn python_rns_available() -> bool {
Command::new("python3")
.args(["-c", "import RNS; print('ok')"])
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|status| status.success())
.unwrap_or(false)
}
fn parse_hex_16(input: &str) -> [u8; 16] {
assert_eq!(input.len(), 32, "expected 16-byte hex string");
let mut out = [0u8; 16];
for i in 0..16 {
out[i] = u8::from_str_radix(&input[i * 2..i * 2 + 2], 16).unwrap();
}
out
}
fn client_rns_config(port: u16) -> String {
format!(
r#"[reticulum]
enable_transport = No
share_instance = No
panic_on_interface_error = Yes
[interfaces]
[[Python RNGit Test TCP]]
type = TCPClientInterface
target_host = 127.0.0.1
target_port = {port}
mode = full
"#
)
}
const PYTHON_RNGIT_PEER_SCRIPT: &str = r#"
import os
import signal
import socket
import sys
import tempfile
import threading
import time
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
sock.close()
config_dir = tempfile.mkdtemp()
with open(os.path.join(config_dir, "config"), "w") as f:
f.write(f"""[reticulum]
enable_transport = false
share_instance = no
[interfaces]
[[TCP Server Interface]]
type = TCPServerInterface
interface_enabled = true
listen_ip = 127.0.0.1
listen_port = {port}
mode = full
""")
import RNS
running = True
require_reidentify = os.environ.get("RNGIT_TEST_REQUIRE_REIDENTIFY") == "1"
identifications = 0
requests = 0
reticulum = RNS.Reticulum(configdir=config_dir)
identity = RNS.Identity()
destination = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "git", "repositories")
def emit(*parts):
print(" ".join(str(part) for part in parts), flush=True)
def list_refs(path, data, request_id, link_id, remote_identity, requested_at):
global requests
requests += 1
remote = remote_identity.hash.hex() if remote_identity is not None else "none"
emit("REQUEST", path, remote, requests, identifications)
if remote_identity is None:
return b"\x01Not identified"
if require_reidentify and identifications < 2:
return b"\x01Not identified"
return b"\x001111111111111111111111111111111111111111 refs/heads/main\n"
def identified(link, remote_identity):
global identifications
identifications += 1
emit("IDENTIFIED", remote_identity.hash.hex(), identifications)
def link_established(link):
emit("LINK_ESTABLISHED")
link.set_remote_identified_callback(identified)
destination.register_request_handler("/git/list", response_generator=list_refs, allow=RNS.Destination.ALLOW_ALL)
destination.set_link_established_callback(link_established)
def announce_loop():
while running:
destination.announce()
time.sleep(0.25)
threading.Thread(target=announce_loop, daemon=True).start()
emit("READY", port, destination.hash.hex())
signal.signal(signal.SIGTERM, lambda *a: sys.exit(0))
try:
for line in sys.stdin:
if line.strip() == "stop":
break
finally:
running = False
"#;
}