add_user, list_users from CLI

pull/19/head
Niko PLP 1 year ago
parent 0c98b2bdc6
commit b9764e7983
  1. 45
      Cargo.lock
  2. 20
      ng-app/src-tauri/src/lib.rs
  3. 28
      ng-sdk-js/src/lib.rs
  4. 42
      ng-wallet/src/lib.rs
  5. 2
      ngaccount/src/main.rs
  6. 16
      ngcli/Cargo.toml
  7. 977
      ngcli/src/main.rs
  8. 587
      ngcli/src/old.rs
  9. 4
      ngd/src/cli.rs
  10. 3
      ngd/src/main.rs
  11. 19
      p2p-broker/src/broker_store/account.rs
  12. 11
      p2p-broker/src/server_ws.rs
  13. 19
      p2p-broker/src/storage.rs
  14. 10
      p2p-broker/src/types.rs
  15. 1
      p2p-client-ws/Cargo.toml
  16. 38
      p2p-client-ws/src/lib.rs
  17. 31
      p2p-client-ws/src/remote_ws.rs
  18. 25
      p2p-client-ws/src/remote_ws_wasm.rs
  19. 9
      p2p-net/src/actor.rs
  20. 109
      p2p-net/src/actors/add_user.rs
  21. 86
      p2p-net/src/actors/del_user.rs
  22. 91
      p2p-net/src/actors/list_users.rs
  23. 9
      p2p-net/src/actors/mod.rs
  24. 18
      p2p-net/src/actors/start.rs
  25. 171
      p2p-net/src/broker.rs
  26. 10
      p2p-net/src/broker_storage.rs
  27. 286
      p2p-net/src/connection.rs
  28. 184
      p2p-net/src/errors.rs
  29. 683
      p2p-net/src/types.rs
  30. 7
      p2p-repo/src/kcv_store.rs
  31. 24
      p2p-repo/src/types.rs
  32. 2
      stores-lmdb/Cargo.toml
  33. 67
      stores-lmdb/src/kcv_store.rs

45
Cargo.lock generated

@ -764,6 +764,7 @@ dependencies = [
"anstyle",
"bitflags",
"clap_lex",
"once_cell",
"strsim",
]
@ -2816,18 +2817,28 @@ dependencies = [
name = "ngcli"
version = "0.1.0"
dependencies = [
"anyhow",
"assert_cmd",
"async-std",
"base64-url",
"blake3",
"clap",
"ed25519-dalek",
"fastbloom-rs",
"env_logger",
"futures",
"p2p-broker",
"getrandom 0.2.10",
"log",
"p2p-client-ws",
"p2p-net",
"p2p-repo",
"rand 0.7.3",
"serde",
"serde_bare",
"serde_bytes",
"serde_json",
"stores-lmdb",
"tempfile",
"zeroize",
]
[[package]]
@ -3135,7 +3146,6 @@ dependencies = [
"wasm-bindgen",
"wasm-bindgen-test",
"ws_stream_wasm",
"xactor",
]
[[package]]
@ -3854,7 +3864,7 @@ dependencies = [
[[package]]
name = "rkv"
version = "0.18.0"
source = "git+https://git.nextgraph.org/NextGraph/rkv.git?rev=8f5ad79c0c93138b1bdc0a1254a7c6b4d357a5d9#8f5ad79c0c93138b1bdc0a1254a7c6b4d357a5d9"
source = "git+https://git.nextgraph.org/NextGraph/rkv.git?rev=c746abb443b7bb4541ebbef2b71e8d0f9eb39f6a#c746abb443b7bb4541ebbef2b71e8d0f9eb39f6a"
dependencies = [
"arrayref",
"bincode",
@ -5922,33 +5932,6 @@ dependencies = [
"zeroize",
]
[[package]]
name = "xactor"
version = "0.7.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a5297548253bd91993bedcc70ae166e64b3fab5ded1965055b2a450777870ff"
dependencies = [
"anyhow",
"async-std",
"async-trait",
"fnv",
"futures",
"once_cell",
"slab",
"xactor-derive",
]
[[package]]
name = "xactor-derive"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce65a6e78fedf06f36b8e7b3154175226fad605971f535062339c05f5caecbed"
dependencies = [
"proc-macro2",
"quote",
"syn 1.0.109",
]
[[package]]
name = "xorfilter-rs"
version = "0.5.1"

@ -31,13 +31,13 @@ pub type SetupHook = Box<dyn FnOnce(&mut App) -> Result<(), Box<dyn std::error::
#[tauri::command(rename_all = "snake_case")]
async fn test() -> Result<(), ()> {
log_info!("test is {}", BROKER.read().await.test());
log_debug!("test is {}", BROKER.read().await.test());
Ok(())
}
#[tauri::command(rename_all = "snake_case")]
async fn wallet_gen_shuffle_for_pazzle_opening(pazzle_length: u8) -> Result<ShuffledPazzle, ()> {
log_info!(
log_debug!(
"wallet_gen_shuffle_for_pazzle_opening from rust {}",
pazzle_length
);
@ -46,7 +46,7 @@ async fn wallet_gen_shuffle_for_pazzle_opening(pazzle_length: u8) -> Result<Shuf
#[tauri::command(rename_all = "snake_case")]
async fn wallet_gen_shuffle_for_pin() -> Result<Vec<u8>, ()> {
log_info!("wallet_gen_shuffle_for_pin from rust");
log_debug!("wallet_gen_shuffle_for_pin from rust");
Ok(gen_shuffle_for_pin())
}
@ -56,13 +56,13 @@ async fn wallet_open_wallet_with_pazzle(
pazzle: Vec<u8>,
pin: [u8; 4],
) -> Result<EncryptedWallet, String> {
log_info!("wallet_open_wallet_with_pazzle from rust {:?}", pazzle);
log_debug!("wallet_open_wallet_with_pazzle from rust {:?}", pazzle);
open_wallet_with_pazzle(wallet, pazzle, pin).map_err(|e| e.to_string())
}
#[tauri::command(rename_all = "snake_case")]
async fn wallet_create_wallet(mut params: CreateWalletV0) -> Result<CreateWalletResultV0, String> {
//log_info!("wallet_create_wallet from rust {:?}", params);
//log_debug!("wallet_create_wallet from rust {:?}", params);
params.result_with_wallet_file = false;
let local_save = params.local_save;
let res = create_wallet_v0(params).await.map_err(|e| e.to_string());
@ -78,13 +78,13 @@ async fn wallet_create_wallet(mut params: CreateWalletV0) -> Result<CreateWallet
#[tauri::command(rename_all = "snake_case")]
async fn encode_create_account(payload: CreateAccountBSP) -> Result<String, ()> {
log_info!("{:?}", payload);
log_debug!("{:?}", payload);
payload.encode().ok_or(())
}
#[tauri::command(rename_all = "snake_case")]
async fn doc_sync_branch(nuri: &str, stream_id: &str, app: tauri::AppHandle) -> Result<(), ()> {
log_info!("doc_sync_branch {} {}", nuri, stream_id);
log_debug!("doc_sync_branch {} {}", nuri, stream_id);
let main_window = app.get_window("main").unwrap();
let mut reader;
@ -107,7 +107,7 @@ async fn doc_sync_branch(nuri: &str, stream_id: &str, app: tauri::AppHandle) ->
BROKER.write().await.tauri_stream_cancel(stream_id);
log_info!("END OF LOOP");
log_debug!("END OF LOOP");
Ok(())
}
@ -118,7 +118,7 @@ async fn doc_sync_branch(nuri: &str, stream_id: &str, app: tauri::AppHandle) ->
#[tauri::command(rename_all = "snake_case")]
async fn cancel_doc_sync_branch(stream_id: &str) -> Result<(), ()> {
log_info!("cancel stream {}", stream_id);
log_debug!("cancel stream {}", stream_id);
BROKER
.write()
.await
@ -131,7 +131,7 @@ async fn doc_get_file_from_store_with_object_ref(
nuri: &str,
obj_ref: ObjectRef,
) -> Result<ObjectContent, String> {
log_info!(
log_debug!(
"doc_get_file_from_store_with_object_ref {} {:?}",
nuri,
obj_ref

@ -148,11 +148,11 @@ pub fn client_info() -> ClientInfoV0 {
#[cfg(target_arch = "wasm32")]
#[wasm_bindgen]
pub fn encode_create_account(payload: JsValue) -> JsValue {
log_info!("{:?}", payload);
log_debug!("{:?}", payload);
let create_account = serde_wasm_bindgen::from_value::<CreateAccountBSP>(payload).unwrap();
log_info!("create_account {:?}", create_account);
log_debug!("create_account {:?}", create_account);
let res = create_account.encode();
log_info!("res {:?}", res);
log_debug!("res {:?}", res);
serde_wasm_bindgen::to_value(&res).unwrap()
}
@ -182,7 +182,7 @@ pub fn client_info() -> ClientInfoV0 {
let ua = client_details();
let bowser = Bowser::parse(ua);
//log_info!("{:?}", bowser);
//log_debug!("{:?}", bowser);
let details_string = client_details2(bowser, env!("CARGO_PKG_VERSION").to_string());
@ -200,9 +200,9 @@ pub fn client_info() -> ClientInfoV0 {
#[cfg(target_arch = "wasm32")]
#[wasm_bindgen]
pub async fn test() {
log_info!("test is {}", BROKER.read().await.test());
log_debug!("test is {}", BROKER.read().await.test());
let client_info = client_info();
log_info!("{:?}", client_info);
log_debug!("{:?}", client_info);
}
#[cfg(target_arch = "wasm32")]
@ -213,7 +213,7 @@ pub async fn doc_get_file_from_store_with_object_ref(
) -> Result<JsValue, JsValue> {
let obj_ref = serde_wasm_bindgen::from_value::<ObjectRef>(obj_ref_js).unwrap();
log_info!(
log_debug!(
"doc_get_file {} {:?} {}",
nuri,
obj_ref.id,
@ -271,14 +271,14 @@ pub async fn doc_sync_branch(anuri: String, callback: &js_sys::Function) -> JsVa
Err(_) => {}
}
}
log_info!("END OF LOOP");
log_debug!("END OF LOOP");
Ok(())
}
spawn_and_log_error(inner_task(reader, anuri, callback.clone()));
let cb = Closure::once(move || {
log_info!("close channel");
log_debug!("close channel");
sender.close_channel()
});
//Closure::wrap(Box::new(move |sender| sender.close_channel()) as Box<FnMut(Sender<Commit>)>);
@ -299,7 +299,7 @@ pub async fn probe() {
WS_PORT,
)
.await;
log_info!("broker.probe : {:?}", res);
log_debug!("broker.probe : {:?}", res);
Broker::join_shutdown_with_timeout(std::time::Duration::from_secs(5)).await;
}
@ -315,12 +315,12 @@ pub async fn start() {
//let pub_key = PubKey::Ed25519PubKey(keys.1);
let keys = generate_keypair();
let x_from_ed = keys.1.to_dh_from_ed();
log_info!("Pub from X {}", x_from_ed);
log_debug!("Pub from X {}", x_from_ed);
let (client_priv, client) = generate_keypair();
let (user_priv, user) = generate_keypair();
log_info!("start connecting");
log_debug!("start connecting");
let res = BROKER
.write()
@ -340,7 +340,7 @@ pub async fn start() {
}),
)
.await;
log_info!("broker.connect : {:?}", res);
log_debug!("broker.connect : {:?}", res);
if res.is_err() {
return Ok(());
//panic!("Cannot connect");
@ -352,7 +352,7 @@ pub async fn start() {
async fn timer_close(remote_peer_id: DirectPeerId, user: Option<PubKey>) -> ResultSend<()> {
async move {
sleep!(std::time::Duration::from_secs(3));
log_info!("timeout");
log_debug!("timeout");
BROKER
.write()
.await

@ -213,7 +213,7 @@ pub fn open_wallet_with_pazzle(
)?;
pazzle_key.zeroize();
log_info!(
log_debug!(
"opening of wallet with pazzle took: {} ms",
opening_pazzle.elapsed().as_millis()
);
@ -298,16 +298,16 @@ pub fn display_pazzle(pazzle: &Vec<u8>) -> Vec<String> {
pub fn gen_shuffle_for_pazzle_opening(pazzle_length: u8) -> ShuffledPazzle {
let mut rng = rand::thread_rng();
let mut category_indices: Vec<u8> = (0..pazzle_length).collect();
//log_info!("{:?}", category_indices);
//log_debug!("{:?}", category_indices);
category_indices.shuffle(&mut rng);
//log_info!("{:?}", category_indices);
//log_debug!("{:?}", category_indices);
let mut emoji_indices: Vec<Vec<u8>> = Vec::with_capacity(pazzle_length.into());
for _ in 0..pazzle_length {
let mut idx: Vec<u8> = (0..15).collect();
//log_info!("{:?}", idx);
//log_debug!("{:?}", idx);
idx.shuffle(&mut rng);
//log_info!("{:?}", idx);
//log_debug!("{:?}", idx);
emoji_indices.push(idx)
}
ShuffledPazzle {
@ -319,9 +319,9 @@ pub fn gen_shuffle_for_pazzle_opening(pazzle_length: u8) -> ShuffledPazzle {
pub fn gen_shuffle_for_pin() -> Vec<u8> {
let mut rng = rand::thread_rng();
let mut digits: Vec<u8> = (0..10).collect();
//log_info!("{:?}", digits);
//log_debug!("{:?}", digits);
digits.shuffle(&mut rng);
//log_info!("{:?}", digits);
//log_debug!("{:?}", digits);
digits
}
@ -335,7 +335,7 @@ pub fn gen_shuffle_for_pin() -> Vec<u8> {
// for i in &mut mnemonic {
// *i = choices.chars().nth(ran.gen_range(0, 72)).unwrap();
// }
// log_info!("{}", mnemonic.iter().collect::<String>());
// log_debug!("{}", mnemonic.iter().collect::<String>());
// }
/// creates a Wallet from a pin, a security text and image (with option to send the bootstrap and wallet to nextgraph.one)
@ -559,7 +559,7 @@ pub async fn create_wallet_v0(
// sig,
// });
log_info!(
log_debug!(
"creating of wallet took: {} ms",
creating_pazzle.elapsed().as_millis()
);
@ -600,11 +600,11 @@ mod test {
#[test]
fn test_gen_shuffle() {
let shuffle = gen_shuffle_for_pazzle_opening(9);
log_info!("{:?}", shuffle);
log_debug!("{:?}", shuffle);
let shuffle = gen_shuffle_for_pazzle_opening(12);
log_info!("{:?}", shuffle);
log_debug!("{:?}", shuffle);
let shuffle = gen_shuffle_for_pazzle_opening(15);
log_info!("{:?}", shuffle);
log_debug!("{:?}", shuffle);
let digits = gen_shuffle_for_pin();
let digits = gen_shuffle_for_pin();
}
@ -636,26 +636,26 @@ mod test {
.await
.expect("create_wallet_v0");
log_info!(
log_debug!(
"creation of wallet took: {} ms",
creation.elapsed().as_millis()
);
log_info!("-----------------------------");
log_debug!("-----------------------------");
let mut file = File::create("tests/wallet.ngw").expect("open wallet write file");
let ser_wallet = to_vec(&NgFile::V0(NgFileV0::Wallet(res.wallet.clone()))).unwrap();
file.write_all(&ser_wallet);
log_info!(
log_debug!(
"wallet id: {:?}",
base64_url::encode(&res.wallet.id().slice())
);
log_info!("pazzle {:?}", display_pazzle(&res.pazzle));
log_info!("mnemonic {:?}", display_mnemonic(&res.mnemonic));
log_info!("pin {:?}", pin);
log_debug!("pazzle {:?}", display_pazzle(&res.pazzle));
log_debug!("mnemonic {:?}", display_mnemonic(&res.mnemonic));
log_debug!("pin {:?}", pin);
if let Wallet::V0(v0) = &res.wallet {
log_info!("security text: {:?}", v0.content.security_txt);
log_debug!("security text: {:?}", v0.content.security_txt);
let mut file =
File::create("tests/generated_security_image.jpg").expect("open write file");
@ -678,7 +678,7 @@ mod test {
.expect("open with mnemonic");
//log_debug!("encrypted part {:?}", w);
log_info!(
log_debug!(
"opening of wallet with mnemonic took: {} ms",
opening_mnemonic.elapsed().as_millis()
);
@ -687,7 +687,7 @@ mod test {
let opening_pazzle = Instant::now();
let w = open_wallet_with_pazzle(Wallet::V0(v0.clone()), res.pazzle.clone(), pin)
.expect("open with pazzle");
log_info!(
log_debug!(
"opening of wallet with pazzle took: {} ms",
opening_pazzle.elapsed().as_millis()
);

@ -75,7 +75,7 @@ async fn main() -> anyhow::Result<()> {
let addr: Vec<&str> = server_address.split(',').collect();
if addr.len() != 3 {
return Err(anyhow!(
"NG_ACCOUNT_SERVER is invalid. format is IP,PORT,PEERID"
"NG_ACCOUNT_SERVER is invalid. format is IP,PORT,PEER_ID"
));
}
let ip: IP = addr[0].into();

@ -8,15 +8,25 @@ description = "CLI command-line interpreter of NextGraph"
repository = "https://git.nextgraph.org/NextGraph/nextgraph-rs"
[dependencies]
p2p-repo = { path = "../p2p-repo" }
p2p-repo = { path = "../p2p-repo", features = ["server_log_output"] }
p2p-net = { path = "../p2p-net" }
p2p-client-ws = { path = "../p2p-client-ws" }
p2p-broker = { path = "../p2p-broker" }
stores-lmdb = { path = "../stores-lmdb" }
async-std = { version = "1.12.0", features = ["attributes"] }
futures = "0.3.24"
tempfile = "3"
fastbloom-rs = "0.5.3"
rand = "0.7"
ed25519-dalek = "1.0.1"
assert_cmd = "2.0.5"
clap = { version = "4.3.4", features = ["env","string","cargo"] }
log = "0.4"
env_logger = "0.10"
anyhow = "1.0.71"
serde_json = "1.0"
zeroize = { version = "1.6.0" }
base64-url = "2.0.0"
getrandom = "0.2.7"
blake3 = "1.3.1"
serde = { version = "1.0", features = ["derive"] }
serde_bare = "0.5.0"
serde_bytes = "0.11.7"

File diff suppressed because it is too large Load Diff

@ -0,0 +1,587 @@
fn block_size() -> usize {
store_max_value_size()
//store_valid_value_size(0)
}
async fn test_sync(cnx: &mut impl BrokerConnection, user_pub_key: PubKey, userpriv_key: PrivKey) {
fn add_obj(
content: ObjectContent,
deps: Vec<ObjectId>,
expiry: Option<Timestamp>,
repo_pubkey: PubKey,
repo_secret: SymKey,
store: &mut impl RepoStore,
) -> ObjectRef {
let max_object_size = 4000;
let obj = Object::new(
content,
deps,
expiry,
max_object_size,
repo_pubkey,
repo_secret,
);
//log_debug!(">>> add_obj");
log_debug!(" id: {}", obj.id());
//log_debug!(" deps: {:?}", obj.deps());
obj.save(store).unwrap();
obj.reference().unwrap()
}
fn add_commit(
branch: ObjectRef,
author_privkey: PrivKey,
author_pubkey: PubKey,
seq: u32,
deps: Vec<ObjectRef>,
acks: Vec<ObjectRef>,
body_ref: ObjectRef,
repo_pubkey: PubKey,
repo_secret: SymKey,
store: &mut impl RepoStore,
) -> ObjectRef {
let mut obj_deps: Vec<ObjectId> = vec![];
obj_deps.extend(deps.iter().map(|r| r.id));
obj_deps.extend(acks.iter().map(|r| r.id));
let obj_ref = ObjectRef {
id: ObjectId::Blake3Digest32([1; 32]),
key: SymKey::ChaCha20Key([2; 32]),
};
let refs = vec![obj_ref];
let metadata = vec![5u8; 55];
let expiry = None;
let commit = Commit::new(
author_privkey,
author_pubkey,
seq,
branch,
deps,
acks,
refs,
metadata,
body_ref,
expiry,
)
.unwrap();
//log_debug!("commit: {}", commit.id().unwrap());
add_obj(
ObjectContent::Commit(commit),
obj_deps,
expiry,
repo_pubkey,
repo_secret,
store,
)
}
fn add_body_branch(
branch: Branch,
repo_pubkey: PubKey,
repo_secret: SymKey,
store: &mut impl RepoStore,
) -> ObjectRef {
let deps = vec![];
let expiry = None;
let body = CommitBody::Branch(branch);
//log_debug!("body: {:?}", body);
add_obj(
ObjectContent::CommitBody(body),
deps,
expiry,
repo_pubkey,
repo_secret,
store,
)
}
fn add_body_trans(
deps: Vec<ObjectId>,
repo_pubkey: PubKey,
repo_secret: SymKey,
store: &mut impl RepoStore,
) -> ObjectRef {
let expiry = None;
let content = [7u8; 777].to_vec();
let body = CommitBody::Transaction(Transaction::V0(content));
//log_debug!("body: {:?}", body);
add_obj(
ObjectContent::CommitBody(body),
deps,
expiry,
repo_pubkey,
repo_secret,
store,
)
}
fn add_body_ack(
deps: Vec<ObjectId>,
repo_pubkey: PubKey,
repo_secret: SymKey,
store: &mut impl RepoStore,
) -> ObjectRef {
let expiry = None;
let body = CommitBody::Ack(Ack::V0());
//log_debug!("body: {:?}", body);
add_obj(
ObjectContent::CommitBody(body),
deps,
expiry,
repo_pubkey,
repo_secret,
store,
)
}
let mut store = HashMapRepoStore::new();
let mut rng = OsRng {};
// repo
let repo_keypair: Keypair = Keypair::generate(&mut rng);
// log_debug!(
// "repo private key: ({}) {:?}",
// repo_keypair.secret.as_bytes().len(),
// repo_keypair.secret.as_bytes()
// );
// log_debug!(
// "repo public key: ({}) {:?}",
// repo_keypair.public.as_bytes().len(),
// repo_keypair.public.as_bytes()
// );
let _repo_privkey = PrivKey::Ed25519PrivKey(repo_keypair.secret.to_bytes());
let repo_pubkey = PubKey::Ed25519PubKey(repo_keypair.public.to_bytes());
let repo_secret = SymKey::ChaCha20Key([9; 32]);
let repolink = RepoLink::V0(RepoLinkV0 {
id: repo_pubkey,
secret: repo_secret,
peers: vec![],
});
// branch
let branch_keypair: Keypair = Keypair::generate(&mut rng);
//log_debug!("branch public key: {:?}", branch_keypair.public.as_bytes());
let branch_pubkey = PubKey::Ed25519PubKey(branch_keypair.public.to_bytes());
let member_keypair: Keypair = Keypair::generate(&mut rng);
//log_debug!("member public key: {:?}", member_keypair.public.as_bytes());
let member_privkey = PrivKey::Ed25519PrivKey(member_keypair.secret.to_bytes());
let member_pubkey = PubKey::Ed25519PubKey(member_keypair.public.to_bytes());
let metadata = [66u8; 64].to_vec();
let commit_types = vec![CommitType::Ack, CommitType::Transaction];
let secret = SymKey::ChaCha20Key([0; 32]);
let member = MemberV0::new(member_pubkey, commit_types, metadata.clone());
let members = vec![member];
let mut quorum = HashMap::new();
quorum.insert(CommitType::Transaction, 3);
let ack_delay = RelTime::Minutes(3);
let tags = [99u8; 32].to_vec();
let branch = Branch::new(
branch_pubkey,
branch_pubkey,
secret,
members,
quorum,
ack_delay,
tags,
metadata,
);
//log_debug!("branch: {:?}", branch);
log_debug!("branch deps/acks:");
log_debug!("");
log_debug!(" br");
log_debug!(" / \\");
log_debug!(" t1 t2");
log_debug!(" / \\ / \\");
log_debug!(" a3 t4<--t5-->(t1)");
log_debug!(" / \\");
log_debug!(" a6 a7");
log_debug!("");
// commit bodies
let branch_body = add_body_branch(
branch.clone(),
repo_pubkey.clone(),
repo_secret.clone(),
&mut store,
);
let ack_body = add_body_ack(vec![], repo_pubkey, repo_secret, &mut store);
let trans_body = add_body_trans(vec![], repo_pubkey, repo_secret, &mut store);
// create & add commits to store
log_debug!(">> br");
let br = add_commit(
branch_body,
member_privkey,
member_pubkey,
0,
vec![],
vec![],
branch_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> t1");
let t1 = add_commit(
branch_body,
member_privkey,
member_pubkey,
1,
vec![br],
vec![],
trans_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> t2");
let t2 = add_commit(
branch_body,
member_privkey,
member_pubkey,
2,
vec![br],
vec![],
trans_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> a3");
let a3 = add_commit(
branch_body,
member_privkey,
member_pubkey,
3,
vec![t1],
vec![],
ack_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> t4");
let t4 = add_commit(
branch_body,
member_privkey,
member_pubkey,
4,
vec![t2],
vec![t1],
trans_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> t5");
let t5 = add_commit(
branch_body,
member_privkey,
member_pubkey,
5,
vec![t1, t2],
vec![t4],
trans_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> a6");
let a6 = add_commit(
branch_body,
member_privkey,
member_pubkey,
6,
vec![t4],
vec![],
ack_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!(">> a7");
let a7 = add_commit(
branch_body,
member_privkey,
member_pubkey,
7,
vec![t4],
vec![],
ack_body,
repo_pubkey,
repo_secret,
&mut store,
);
let mut public_overlay_cnx = cnx
.overlay_connect(&repolink, true)
.await
.expect("overlay_connect failed");
// Sending everything to the broker
for (v) in store.get_all() {
//log_debug!("SENDING {}", k);
let _ = public_overlay_cnx
.put_block(&v)
.await
.expect("put_block failed");
}
// Now emptying the local store of the client, and adding only 1 commit into it (br)
// we also have received an commit (t5) but we don't know what to do with it...
let mut store = HashMapRepoStore::new();
let br = add_commit(
branch_body,
member_privkey,
member_pubkey,
0,
vec![],
vec![],
branch_body,
repo_pubkey,
repo_secret,
&mut store,
);
let t5 = add_commit(
branch_body,
member_privkey,
member_pubkey,
5,
vec![t1, t2],
vec![t4],
trans_body,
repo_pubkey,
repo_secret,
&mut store,
);
log_debug!("LOCAL STORE HAS {} BLOCKS", store.get_len());
// Let's pretend that we know that the head of the branch in the broker is at commits a6 and a7.
// normally it would be the pub/sub that notifies us of those heads.
// now we want to synchronize with the broker.
let mut filter = Filter::new(FilterBuilder::new(10, 0.01));
for commit_ref in [br, t5] {
match commit_ref.id {
ObjectId::Blake3Digest32(d) => filter.add(&d),
}
}
let cfg = filter.config();
let known_commits = BloomFilter {
k: cfg.hashes,
f: filter.get_u8_array().to_vec(),
};
let known_heads = [br.id];
let remote_heads = [a6.id, a7.id];
let mut synced_blocks_stream = public_overlay_cnx
.sync_branch(remote_heads.to_vec(), known_heads.to_vec(), known_commits)
.await
.expect("sync_branch failed");
let mut i = 0;
while let Some(b) = synced_blocks_stream.next().await {
log_debug!("GOT BLOCK {}", b.id());
store.put(&b);
i += 1;
}
log_debug!("SYNCED {} BLOCKS", i);
log_debug!("LOCAL STORE HAS {} BLOCKS", store.get_len());
// now the client can verify the DAG and each commit. Then update its list of heads.
}
async fn test(
cnx: &mut impl BrokerConnection,
pub_key: PubKey,
priv_key: PrivKey,
) -> Result<(), ProtocolError> {
cnx.add_user(PubKey::Ed25519PubKey([1; 32]), priv_key)
.await?;
cnx.add_user(pub_key, priv_key).await?;
//.expect("add_user 2 (myself) failed");
assert_eq!(
cnx.add_user(PubKey::Ed25519PubKey([1; 32]), priv_key)
.await
.err()
.unwrap(),
ProtocolError::UserAlreadyExists
);
let repo = RepoLink::V0(RepoLinkV0 {
id: PubKey::Ed25519PubKey([1; 32]),
secret: SymKey::ChaCha20Key([0; 32]),
peers: vec![],
});
let mut public_overlay_cnx = cnx.overlay_connect(&repo, true).await?;
log_debug!("put_block");
let my_block_id = public_overlay_cnx
.put_block(&Block::new(
vec![],
ObjectDeps::ObjectIdList(vec![]),
None,
vec![27; 150],
None,
))
.await?;
log_debug!("added block_id to store {}", my_block_id);
let object_id = public_overlay_cnx
.put_object(
ObjectContent::File(File::V0(FileV0 {
content_type: vec![],
metadata: vec![],
content: vec![48; 69000],
})),
vec![],
None,
block_size(),
repo.id(),
repo.secret(),
)
.await?;
log_debug!("added object_id to store {}", object_id);
let mut my_block_stream = public_overlay_cnx
.get_block(my_block_id, true, None)
.await?;
//.expect("get_block failed");
while let Some(b) = my_block_stream.next().await {
log_debug!("GOT BLOCK {}", b.id());
}
let mut my_object_stream = public_overlay_cnx.get_block(object_id, true, None).await?;
//.expect("get_block for object failed");
while let Some(b) = my_object_stream.next().await {
log_debug!("GOT BLOCK {}", b.id());
}
let object = public_overlay_cnx.get_object(object_id, None).await?;
//.expect("get_object failed");
log_debug!("GOT OBJECT with ID {}", object.id());
// let object_id = public_overlay_cnx
// .copy_object(object_id, Some(now_timestamp() + 60))
// .await
// .expect("copy_object failed");
// log_debug!("COPIED OBJECT to OBJECT ID {}", object_id);
public_overlay_cnx.delete_object(object_id).await?;
//.expect("delete_object failed");
let res = public_overlay_cnx
.get_object(object_id, None)
.await
.unwrap_err();
log_debug!("result from get object after delete: {}", res);
assert_eq!(res, ProtocolError::NotFound);
//TODO test pin/unpin
// TEST BRANCH SYNC
test_sync(cnx, pub_key, priv_key).await;
Ok(())
}
async fn test_local_connection() {
log_debug!("===== TESTING LOCAL API =====");
let root = tempfile::Builder::new().prefix("ngcli").tempdir().unwrap();
let master_key: [u8; 32] = [0; 32];
std::fs::create_dir_all(root.path()).unwrap();
log_debug!("{}", root.path().to_str().unwrap());
let store = LmdbKCVStore::open(root.path(), master_key);
//let mut server = BrokerServer::new(store, ConfigMode::Local).expect("starting broker");
let (priv_key, pub_key) = generate_keypair();
// let mut cnx = server.local_connection(pub_key);
// test(&mut cnx, pub_key, priv_key).await;
}
async fn test_remote_connection(url: &str) {
log_debug!("===== TESTING REMOTE API =====");
let (priv_key, pub_key) = generate_keypair();
// open cnx
// test(&mut cnx, pub_key, priv_key).await;
}
#[cfg(test)]
mod test {
use crate::{test_local_connection, test_remote_connection};
#[async_std::test]
pub async fn test_local_cnx() {}
use async_std::task;
use p2p_broker::server_ws::*;
use p2p_net::utils::gen_dh_keys;
use p2p_net::WS_PORT;
use p2p_repo::log::*;
use p2p_repo::types::PubKey;
#[async_std::test]
pub async fn test_remote_cnx() -> Result<(), Box<dyn std::error::Error>> {
let keys = gen_dh_keys();
// log_debug!("Public key of node: {:?}", keys.1);
// log_debug!("Private key of node: {:?}", keys.0.as_slice());
log_debug!("Public key of node: {}", keys.1);
log_debug!("Private key of node: {}", keys.0);
let thr = task::spawn(run_server_accept_one("127.0.0.1", WS_PORT, keys.0, pubkey));
// time for the server to start
std::thread::sleep(std::time::Duration::from_secs(2));
test_remote_connection("ws://127.0.0.1:3012");
thr.await;
Ok(())
}
}

@ -31,7 +31,7 @@ pub(crate) struct Cli {
#[arg(short, long, env = "NG_SERVER_KEY")]
pub key: Option<String>,
/// Saves to disk the provided or automatically generated key. Only used if file storage is secure. Alternatives are passing the key at every start with --key or NG_SERVER_KEY env var.
/// Saves to disk the provided or automatically generated key. Only use if file storage is secure. Alternatives are passing the key at every start with --key or NG_SERVER_KEY env var.
#[arg(long)]
pub save_key: bool,
@ -47,7 +47,7 @@ pub(crate) struct Cli {
#[arg(long, requires("core"))]
pub core_with_clients: bool,
/// Quick config to forward all requests to another BROKER. format is "[DOMAIN/IP:PORT]@PEERID". An IPv6 should be encased in square brackets [IPv6] and the whole option should be between double quotes. Port defaults to 80 for IPs and 443 for domains
/// Quick config to forward all requests to another BROKER. format is "[DOMAIN/IP:PORT]@PEER_ID". An IPv6 should be encased in square brackets [IPv6] and the whole option should be between double quotes. Port defaults to 80 for IPs and 443 for domains
#[arg(
short,
long,

@ -450,6 +450,7 @@ async fn main_inner() -> Result<(), ()> {
|| args.public.is_some()
|| args.dynamic.is_some()
|| args.domain.is_some()
|| args.domain_private.is_some()
{
// QUICK CONFIG
@ -869,7 +870,7 @@ async fn main_inner() -> Result<(), ()> {
return Err(());
}
let pub_key_array = decode_key(parts[1])
.map_err(|_| log_err!("The PEERID provided in the --forward option is invalid"))?;
.map_err(|_| log_err!("The PEER_ID provided in the --forward option is invalid"))?;
let peer_id = PubKey::Ed25519PubKey(pub_key_array);
let server_type = if parts[0].len() > 0 {

@ -18,7 +18,7 @@ use p2p_net::types::*;
use p2p_repo::kcv_store::KCVStore;
use p2p_repo::store::*;
use p2p_repo::types::Timestamp;
use serde_bare::to_vec;
use serde_bare::{from_slice, to_vec};
pub struct Account<'a> {
/// User ID
@ -65,7 +65,7 @@ impl<'a> Account<'a> {
store,
};
if acc.exists() {
return Err(StorageError::BackendError);
return Err(StorageError::AlreadyExists);
}
store.put(
Self::PREFIX,
@ -75,6 +75,21 @@ impl<'a> Account<'a> {
)?;
Ok(acc)
}
pub fn get_all_users(
admins: bool,
store: &'a dyn KCVStore,
) -> Result<Vec<UserId>, StorageError> {
let size = to_vec(&UserId::nil())?.len();
let mut res: Vec<UserId> = vec![];
for user in store.get_all_keys_and_values(Self::PREFIX, size, Some(Self::ADMIN))? {
let admin: bool = from_slice(&user.1)?;
if admin == admins {
let id: UserId = from_slice(&user.0[1..user.0.len() - 1])?;
res.push(id);
}
}
Ok(res)
}
pub fn exists(&self) -> bool {
self.store
.get(

@ -595,7 +595,7 @@ pub async fn run_server_accept_one(
let tcp = connections.next().await.unwrap()?;
{
BROKER.write().await.set_my_peer_id(peer_pub_key);
//BROKER.write().await.set_my_peer_id(peer_pub_key);
}
accept(tcp, peer_priv_key).await;
@ -779,12 +779,17 @@ pub async fn run_server_v0(
.map_err(|e| log_err!("Error while opening broker storage: {:?}", e))?;
let mut broker = BROKER.write().await;
broker.set_my_peer_id(peer_id);
broker.set_storage(broker_storage);
LISTENERS_INFO
.set(broker.set_listeners(listener_infos))
.unwrap();
broker.set_overlays_configs(config.overlays_configs);
let server_config = ServerConfig {
overlays_configs: config.overlays_configs,
registration: config.registration,
admin_user: config.admin_user,
peer_id,
};
broker.set_server_config(server_config);
}
// Actually starting the listeners

@ -11,18 +11,21 @@
use std::path::PathBuf;
use crate::broker_store::account::Account;
use crate::broker_store::invitation::Invitation;
use crate::broker_store::wallet::Wallet;
use crate::types::*;
use p2p_net::broker_storage::*;
use p2p_net::errors::ProtocolError;
use p2p_net::types::{BootstrapContentV0, InvitationCode, InvitationV0};
use p2p_repo::kcv_store::KCVStore;
use p2p_repo::log::*;
use p2p_repo::store::StorageError;
use p2p_repo::types::SymKey;
use p2p_repo::types::{PubKey, SymKey};
use stores_lmdb::kcv_store::LmdbKCVStore;
use stores_lmdb::repo_store::LmdbRepoStore;
#[derive(Debug)]
pub struct LmdbBrokerStorage {
wallet_storage: LmdbKCVStore,
accounts_storage: LmdbKCVStore,
@ -94,5 +97,17 @@ impl LmdbBrokerStorage {
}
impl BrokerStorage for LmdbBrokerStorage {
fn get_user(&self) {}
fn get_user(&self, user_id: PubKey) -> Result<bool, ProtocolError> {
log_debug!("get_user {user_id}");
Ok(Account::open(&user_id, &self.accounts_storage)?.is_admin()?)
}
fn add_user(&self, user_id: PubKey, is_admin: bool) -> Result<(), ProtocolError> {
log_debug!("add_user {user_id} is admin {is_admin}");
Account::create(&user_id, is_admin, &self.accounts_storage)?;
Ok(())
}
fn list_users(&self, admins: bool) -> Result<Vec<PubKey>, ProtocolError> {
log_debug!("list_users that are admin == {admins}");
Ok(Account::get_all_users(admins, &self.accounts_storage)?)
}
}

@ -6,18 +6,10 @@
// at your option. All files in the project carrying such
// notice may not be copied, modified, or distributed except
// according to those terms.
use p2p_net::types::{BrokerOverlayConfigV0, ListenerV0};
use p2p_net::types::{BrokerOverlayConfigV0, ListenerV0, RegistrationConfig};
use p2p_repo::types::{PrivKey, PubKey};
use serde::{Deserialize, Serialize};
/// Registration config
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum RegistrationConfig {
Closed,
Invitation,
Open,
}
/// DaemonConfig Version 0
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DaemonConfigV0 {

@ -33,5 +33,4 @@ features = ["js"]
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
getrandom = "0.2.7"
xactor = "0.7.11"
async-tungstenite = { git = "https://git.nextgraph.org/NextGraph/async-tungstenite.git", branch = "nextgraph", features = ["async-std-runtime"] }

@ -6,44 +6,6 @@
// notice may not be copied, modified, or distributed except
// according to those terms.
#[macro_export]
macro_rules! before {
( $self:expr, $request_id:ident, $addr:ident, $receiver:ident ) => {
let mut actor = BrokerMessageActor::new();
let $receiver = actor.receiver();
let mut $addr = actor
.start()
.await
.map_err(|_e| ProtocolError::ActorError)?;
let $request_id = $addr.actor_id();
//log_debug!("actor ID {}", $request_id);
{
let mut map = $self.actors.write().expect("RwLock poisoned");
map.insert($request_id, $addr.downgrade());
}
};
}
macro_rules! after {
( $self:expr, $request_id:ident, $addr:ident, $receiver:ident, $reply:ident ) => {
//log_debug!("waiting for reply");