From 0c66cc7393cbb57362b1a9aa5fa8dca1b7f5fe2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Mon, 16 May 2022 12:06:15 +0300 Subject: [PATCH] Feature/service stats (#1267) * Send message from service provider to stats service * Put some actual data in stats * Put stats sender on its own thread and send response data too * Use SQLite for storing stats * Add the data interval and timestamp * Fix clippy * Set description at boot * Guard stats service functionality under a feature for now * Make stats service address data into consts * Add README to network requester * Retrieve sql data in interval * Expose sql data via rocket rest api * Add entry to changelog --- CHANGELOG.md | 2 + Cargo.lock | 6 + clients/credential/credential.db | 1 + clients/socks5/src/socks/client.rs | 8 +- common/socks5/requests/src/lib.rs | 2 + common/socks5/requests/src/msg.rs | 63 ++++++++ .../network-requester/Cargo.toml | 21 ++- service-providers/network-requester/README.md | 26 ++++ service-providers/network-requester/build.rs | 28 ++++ .../20220512120000_mixnet_statistics.sql | 14 ++ .../network-requester/src/core.rs | 131 ++++++++++++++--- .../network-requester/src/main.rs | 34 ++++- .../network-requester/src/statistics/comm.rs | 139 ++++++++++++++++++ .../network-requester/src/statistics/error.rs | 10 ++ .../network-requester/src/statistics/mod.rs | 9 ++ .../network-requester/src/statistics/timer.rs | 40 +++++ .../network-requester/src/storage/error.rs | 14 ++ .../network-requester/src/storage/manager.rs | 62 ++++++++ .../network-requester/src/storage/mod.rs | 92 ++++++++++++ .../network-requester/src/storage/models.rs | 15 ++ .../network-requester/src/storage/routes.rs | 49 ++++++ 21 files changed, 736 insertions(+), 30 deletions(-) create mode 100644 clients/credential/credential.db create mode 100644 common/socks5/requests/src/msg.rs create mode 100644 service-providers/network-requester/README.md create mode 100644 service-providers/network-requester/build.rs create mode 100644 service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql create mode 100644 service-providers/network-requester/src/statistics/comm.rs create mode 100644 service-providers/network-requester/src/statistics/error.rs create mode 100644 service-providers/network-requester/src/statistics/mod.rs create mode 100644 service-providers/network-requester/src/statistics/timer.rs create mode 100644 service-providers/network-requester/src/storage/error.rs create mode 100644 service-providers/network-requester/src/storage/manager.rs create mode 100644 service-providers/network-requester/src/storage/mod.rs create mode 100644 service-providers/network-requester/src/storage/models.rs create mode 100644 service-providers/network-requester/src/storage/routes.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index ac0d253e04..e2e4b58981 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ - mixnet-contract: Replace all naked `-` with `saturating_sub`. - validator-api: add Swagger to document the REST API ([#1249]). - all: added network compilation target to `--help` (or `--version`) commands ([#1256]). +- network-requester: send traffic statistics from all network requesters and receive it in a special network-requester that aggregates the data and exposes it via a rest API ([#1267]). ### Fixed @@ -22,6 +23,7 @@ [#1257]: https://github.com/nymtech/nym/pull/1257 [#1260]: https://github.com/nymtech/nym/pull/1260 [#1265]: https://github.com/nymtech/nym/pull/1265 +[#1267]: https://github.com/nymtech/nym/pull/1267 ## [nym-wallet-v1.0.4](https://github.com/nymtech/nym/tree/nym-wallet-v1.0.4) (2022-05-04) diff --git a/Cargo.lock b/Cargo.lock index d443521d4a..225174005d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3165,18 +3165,23 @@ dependencies = [ name = "nym-network-requester" version = "1.0.1" dependencies = [ + "bincode", "clap 2.34.0", "dirs", "futures", "ipnetwork", "log", + "network-defaults", "nymsphinx", "ordered-buffer", "pretty_env_logger", "proxy-helpers", "publicsuffix", "rand 0.7.3", + "serde", "socks5-requests", + "sqlx", + "thiserror", "tokio", "tokio-tungstenite", "websocket-requests", @@ -5219,6 +5224,7 @@ dependencies = [ "bitflags", "byteorder", "bytes", + "chrono", "crc", "crossbeam-queue", "either", diff --git a/clients/credential/credential.db b/clients/credential/credential.db new file mode 100644 index 0000000000..cdc6e614b3 --- /dev/null +++ b/clients/credential/credential.db @@ -0,0 +1 @@ +[{"1C703C0BF3FC559E09E3DFB9846A9194191915E55A03F999EDC9279319CEBDDB":"{\"amount\":1000000,\"tx_hash\":\"1C703C0BF3FC559E09E3DFB9846A9194191915E55A03F999EDC9279319CEBDDB\",\"signing_keypair\":{\"public_key\":\"CacF8UyTpcxN1eueg8MG4dF5Trf74ZX1joWdGyuCQSYw\",\"private_key\":\"F8rkWjiEeNqXGY2dApxMdF9j15mn6ErhsAfBnmX3AMnX\"},\"encryption_keypair\":{\"public_key\":\"2Gvhyib4CJx3AVb8bg1fThPJ1FdySoapjgW8AzFyqymC\",\"private_key\":\"7WmF61KHWxiauzR4r82fK7NR8V7t5dbp2FrafP9KfYQx\"},\"blind_request_data\":{\"serial_number\":[217,158,131,252,67,46,125,220,192,31,52,16,80,226,101,249,38,236,180,207,127,126,24,218,158,216,7,175,240,195,13,93],\"binding_number\":[190,14,43,219,219,238,38,109,64,129,247,215,26,111,122,225,142,158,111,115,187,225,250,248,43,100,216,89,87,75,229,40],\"first_attribute\":[108,111,200,242,190,67,6,60,28,121,105,77,36,198,104,18,146,110,173,47,3,55,92,195,251,51,155,86,30,2,181,90],\"second_attribute\":[26,54,139,25,107,59,204,245,146,62,178,206,95,198,20,95,44,70,32,212,88,167,204,255,51,158,246,232,121,69,16,51],\"blind_sign_req\":[185,251,71,165,120,153,76,190,174,242,198,198,141,177,212,57,39,196,3,42,196,120,132,7,53,141,132,255,228,157,26,20,99,212,42,191,44,17,36,61,240,224,34,224,94,78,90,138,162,198,53,118,137,234,10,52,230,147,204,10,109,217,71,227,216,255,39,7,208,11,37,176,177,195,165,12,99,177,188,217,201,54,220,215,217,191,251,37,37,24,71,45,140,66,222,236,2,0,0,0,0,0,0,0,175,191,107,206,203,123,5,142,154,151,146,69,203,158,133,118,60,247,48,204,14,167,114,15,129,190,148,183,252,35,82,22,126,151,42,197,43,177,192,193,84,156,68,244,122,228,14,190,145,147,5,150,242,184,228,43,251,20,165,25,97,236,201,237,40,47,157,235,165,92,9,192,34,83,204,136,91,226,97,11,245,8,219,50,22,85,192,169,35,31,118,4,144,9,80,70,222,110,208,77,182,13,80,85,95,165,150,127,48,181,210,180,74,240,173,43,118,215,55,146,218,157,130,108,202,139,154,15,201,88,253,134,208,203,164,215,91,243,24,70,28,129,137,249,47,131,77,91,75,32,37,192,154,13,95,110,91,241,219,113,2,0,0,0,0,0,0,0,73,102,227,42,178,212,22,215,219,83,51,155,135,226,79,50,158,48,34,141,160,239,129,217,207,157,209,74,167,156,90,88,215,245,3,108,147,132,224,225,40,189,35,225,253,180,135,247,45,67,64,217,83,75,160,14,207,75,116,210,220,192,189,3,2,0,0,0,0,0,0,0,18,196,183,38,67,8,219,14,183,60,138,123,31,55,31,18,151,74,184,0,61,84,35,173,192,178,50,117,229,225,148,6,104,98,54,139,122,40,129,216,190,185,184,135,121,241,170,207,59,190,203,94,212,222,34,27,160,152,233,52,241,175,84,67]},\"signature\":\"y59zgkBne96tAKREkDQ5swA4mCimpooVLyLNAL3jzdVXVHHd43SoS56pfnHULzs9yHP3AJnGCR6FLyHKEhBSXPrYfEGSLMdhy9VrjYiyhthmd6BsKWqS9rGAG6PbWt2PpDh\"}","90C9B03B448608B95D13128472789B27399100D18C46A49254CB13AD07A0BE88":"{\"amount\":1000000,\"tx_hash\":\"90C9B03B448608B95D13128472789B27399100D18C46A49254CB13AD07A0BE88\",\"signing_keypair\":{\"public_key\":\"EadiWGJPJozdxSMPK3r1sr8oHZEcVQU5qWUmqAn7jDNQ\",\"private_key\":\"ASQHYmsq4X1eMEuQvFwGiunXuZxEzm79PPydmY6CKWZt\"},\"encryption_keypair\":{\"public_key\":\"3d4jyDrdepHZPRQo6KzQFpR5kvPySqYqWC19H51cjhMh\",\"private_key\":\"HDMhDCTiTu5ChUxWJqZRCR7vZnYULyTu5Rx5Vh2VNxTm\"},\"blind_request_data\":{\"serial_number\":[153,108,141,81,89,232,4,244,22,216,186,112,32,142,11,42,87,74,40,216,191,253,127,49,67,243,183,188,162,46,221,75],\"binding_number\":[7,68,244,43,146,101,216,254,14,27,164,250,41,195,187,242,189,144,126,105,179,35,128,247,30,101,200,151,235,224,92,80],\"first_attribute\":[175,216,242,214,61,233,96,76,17,254,79,233,23,217,84,94,32,68,225,249,201,46,109,183,90,63,126,121,114,87,239,44],\"second_attribute\":[180,82,202,163,28,149,147,178,186,66,26,149,55,177,45,154,185,20,62,68,123,70,188,73,28,146,99,235,91,225,49,107],\"blind_sign_req\":[153,12,234,149,97,21,0,51,175,113,17,215,121,255,21,185,27,59,204,115,208,171,136,219,140,125,151,168,85,240,167,222,13,162,250,39,50,74,172,251,237,85,136,41,34,71,198,215,151,16,80,136,115,178,27,230,102,25,4,12,17,204,225,49,67,109,125,119,100,204,201,90,210,111,118,15,146,244,29,130,64,181,67,178,222,190,27,122,12,47,56,63,187,66,236,130,2,0,0,0,0,0,0,0,161,138,107,198,253,152,7,86,206,134,122,66,171,152,77,90,251,134,64,95,193,174,162,252,176,193,217,52,36,53,207,237,188,84,208,204,183,94,103,168,118,149,151,175,40,134,171,53,184,102,151,180,126,82,124,138,79,130,214,135,133,54,251,182,165,32,6,4,34,75,153,113,11,49,157,149,28,150,8,71,214,47,224,119,109,16,113,85,140,181,214,194,11,168,127,178,189,117,99,79,90,134,207,33,102,42,232,150,89,209,247,165,80,91,237,42,201,223,187,88,98,217,4,199,226,1,60,15,45,159,103,94,185,43,99,142,109,186,212,240,9,161,247,135,184,165,75,67,32,47,122,196,39,199,184,176,87,130,30,20,2,0,0,0,0,0,0,0,117,234,104,216,238,168,141,1,212,50,149,242,215,231,19,91,45,18,164,168,24,68,40,57,25,141,165,63,152,72,121,27,135,89,68,176,9,8,176,145,131,50,160,195,7,248,80,104,135,142,236,251,21,55,39,170,86,133,224,50,39,88,252,31,2,0,0,0,0,0,0,0,65,93,35,204,132,168,2,49,218,27,106,107,95,165,195,206,45,63,158,197,65,106,131,71,43,76,129,106,87,155,146,97,187,218,99,40,125,68,50,152,115,112,80,96,21,11,203,94,118,186,209,236,234,35,159,178,148,59,238,39,218,225,219,106]},\"signature\":\"u3BgWtmZZoNG9tSkC37Z6s15EZXLkY4PiYJX6HHA4iwHQKqKvNVuruPaRaP5ra9448G8iBDbybL94RSRfRYZGbeheMcnwaynvuUJbVh3KmdGLJR3Yr5UoNNjoLqEDLzP3jV\"}","04296C930C81F37096FD180CD93896A164CF5F999DF7D09C0124BE1DC75A5358":"{\"amount\":1000000,\"tx_hash\":\"04296C930C81F37096FD180CD93896A164CF5F999DF7D09C0124BE1DC75A5358\",\"signing_keypair\":{\"public_key\":\"9qnzPdvyUjgWFj7RPpvfbEkG7CbnUUdfj1Zi4nKA8B9b\",\"private_key\":\"ALa8vt73t6sjgbeHg5C6NuezvHbENMabZ3wzxv47TGyo\"},\"encryption_keypair\":{\"public_key\":\"E15DAJTRDvAj7JfZ7Rn5DUffcu72HVJyqYX5gm9WVPZu\",\"private_key\":\"4GLy6J8FVTDRdNzuKyDRg1BRQXNxA8gCALaTJKfYYkNR\"},\"blind_request_data\":{\"serial_number\":[67,28,114,203,240,92,183,145,79,253,109,235,240,221,156,199,216,66,67,175,124,158,226,171,141,131,244,120,187,230,146,62],\"binding_number\":[129,8,63,239,8,211,192,30,124,87,30,110,102,86,77,95,149,127,177,126,28,196,245,29,63,127,128,236,82,231,207,96],\"first_attribute\":[19,21,49,159,55,242,100,60,218,124,120,115,254,64,144,181,222,30,31,41,163,110,117,11,0,11,206,221,126,242,191,41],\"second_attribute\":[29,63,187,178,21,49,68,13,190,62,208,224,249,34,37,116,201,114,139,210,185,67,187,4,41,166,130,243,171,5,10,50],\"blind_sign_req\":[171,30,44,213,50,220,70,49,93,74,79,21,41,89,12,7,20,204,72,50,202,62,231,87,64,154,132,195,48,55,131,212,232,22,174,42,151,232,149,62,132,149,238,112,133,68,224,197,164,96,56,253,196,195,89,4,213,177,122,252,77,167,117,179,245,15,85,100,250,103,77,53,61,61,123,229,206,157,80,36,36,254,220,48,88,110,197,43,117,255,59,58,140,60,99,116,2,0,0,0,0,0,0,0,137,28,145,218,237,220,203,213,63,138,166,104,255,246,227,123,203,7,135,135,17,186,27,83,222,28,234,44,53,189,72,192,238,95,141,136,30,65,57,226,125,137,230,137,165,111,183,73,147,55,3,79,148,223,224,23,198,63,0,58,38,228,229,26,206,77,214,109,208,178,255,168,27,124,121,219,174,252,174,110,71,121,172,24,65,222,31,137,239,55,117,144,153,19,54,107,146,125,29,228,135,153,81,207,193,104,103,17,31,180,67,194,197,205,54,14,225,96,193,141,145,121,219,226,227,196,200,37,110,168,46,27,73,173,238,35,26,59,50,148,24,224,90,88,145,12,206,181,108,216,217,49,19,4,105,249,48,205,167,61,2,0,0,0,0,0,0,0,230,46,74,45,156,42,223,24,154,49,207,38,172,223,75,51,137,46,151,242,155,118,37,115,132,207,113,134,124,78,244,59,181,239,71,132,221,190,90,37,6,50,91,214,82,47,39,151,91,78,249,241,56,242,144,138,9,127,32,51,36,164,155,98,2,0,0,0,0,0,0,0,108,190,2,160,139,55,81,228,126,190,247,27,209,83,87,146,171,103,186,168,4,119,76,165,105,105,67,43,159,106,61,87,221,24,235,97,67,137,80,143,180,205,102,162,244,126,43,144,21,243,251,236,15,174,158,163,180,20,123,72,235,124,31,57]},\"signature\":\"yd9yDABbw8c3wgT8Pof7Z2jidAJs8RoL8KbYH2F1tmrMMCJwhUMzSjpyma6kS9WgaJsFBEbNXpwH1ip6LETfPgvfUJJo2Z9c5312TMAfR8qFr4Zc3X9aQyAo8kMvijCT7jA\"}","5E3CF2741F73091EA9B651B271D9925D6FBE3F67E9C5D3793303DA6D44F12FE5":"{\"amount\":1000000,\"tx_hash\":\"5E3CF2741F73091EA9B651B271D9925D6FBE3F67E9C5D3793303DA6D44F12FE5\",\"signing_keypair\":{\"public_key\":\"AEu3bhLcRZBxjoDVaTmpbNqPhyXRuiN5RQAhoRhcwjxt\",\"private_key\":\"DoEaH8EnoXef7SDyuSTuFt5CMWMdErWrBVfRrpnhmJYq\"},\"encryption_keypair\":{\"public_key\":\"HAW7XdJUMGV5fWAQLpZnHMX7nEnGJsRdgekVgYiqzwPL\",\"private_key\":\"G8wSowRi8weKD2dnM8YBvaGB94FEUrmsxLVQK7x6wz4k\"},\"blind_request_data\":{\"serial_number\":[184,233,37,105,38,174,157,246,96,46,155,173,169,113,202,128,121,35,166,191,146,13,251,46,141,104,88,140,102,165,124,33],\"binding_number\":[46,252,137,11,240,92,139,154,129,213,239,195,94,37,102,119,146,151,182,103,77,171,231,223,13,235,14,240,134,92,29,112],\"first_attribute\":[81,14,113,15,19,75,0,228,7,169,12,125,185,125,247,228,63,164,101,82,90,68,65,89,178,210,131,190,14,66,152,55],\"second_attribute\":[174,224,189,178,99,214,9,196,186,65,150,19,176,132,244,148,52,17,138,132,35,205,216,120,59,138,48,210,149,139,35,51],\"blind_sign_req\":[171,19,145,159,249,120,163,237,106,166,105,120,112,17,203,245,126,95,37,190,221,18,186,211,218,99,1,138,95,12,189,224,120,114,175,30,76,87,106,207,58,154,38,30,109,233,29,16,151,198,168,43,105,60,149,9,167,47,211,231,33,186,239,191,191,118,37,242,211,204,130,120,78,11,51,132,62,148,73,186,225,81,140,67,32,64,90,253,196,89,74,31,1,165,183,22,2,0,0,0,0,0,0,0,167,72,204,0,179,90,227,17,51,115,156,33,198,76,219,218,168,111,226,107,228,150,49,239,252,92,166,195,133,251,3,13,167,75,235,27,132,162,11,238,215,48,32,222,253,156,10,133,148,143,7,75,15,86,228,116,119,12,48,63,189,1,131,94,46,245,82,150,198,204,251,125,76,147,205,235,95,153,177,161,233,11,49,243,106,216,201,6,31,110,185,11,166,171,42,143,5,127,215,168,27,229,207,225,197,236,87,213,24,123,205,174,250,156,178,255,204,108,198,150,53,188,144,14,217,234,92,80,123,101,99,157,7,197,207,108,35,125,251,52,180,185,67,155,220,175,32,116,202,128,93,92,77,95,79,205,62,99,110,13,2,0,0,0,0,0,0,0,10,126,251,84,164,181,206,92,82,231,45,58,6,5,79,13,255,73,78,157,121,48,109,118,27,225,108,237,11,228,105,78,177,226,89,24,13,31,52,160,195,224,191,232,115,107,242,8,173,82,83,32,37,118,141,172,217,238,110,243,141,174,211,62,2,0,0,0,0,0,0,0,163,242,24,184,82,249,187,221,4,105,16,203,144,106,58,115,50,23,107,30,252,71,101,155,37,5,227,23,122,10,252,65,18,113,129,88,56,199,169,202,73,196,151,1,186,168,155,7,68,165,186,157,18,222,194,25,194,58,49,125,114,43,212,4]},\"signature\":\"uHR5LkaHQc3LQUXuzydXLj783vsJtp2s5FV1CNRQ3g8rjL29U2UM3wSEFWrxjoFVq5gS8sNUEDbvhGTehWM3pGPRMVY4U68gkPWzzR7dAMX1LHYjwUdUiwNgURmgVFrG7XV\"}","3A24DE7C3C1EF38F6CBEBB1BE893EB6819E55A9C8E8164C55A9B75CE42950536":"{\"amount\":1000000,\"tx_hash\":\"3A24DE7C3C1EF38F6CBEBB1BE893EB6819E55A9C8E8164C55A9B75CE42950536\",\"signing_keypair\":{\"public_key\":\"91wBGhrPnLAM7rc8C1WK693VwfZKj7d9K7Rpfcpxo3x7\",\"private_key\":\"FEh9rCDMZx29nZwTyjezxtcytRiZZrsxM2qWdyMyuEC7\"},\"encryption_keypair\":{\"public_key\":\"9Ay88XFrErtSLeeD3DQETeaqrAgAKNR1vNjoe2PvzyoD\",\"private_key\":\"HD92abV3ZPW1VKxh3bCcHVZ8oJ5XqRjrKspuZkQMvqkT\"},\"blind_request_data\":{\"serial_number\":[132,225,105,217,220,29,178,224,94,120,11,53,170,48,254,118,226,13,138,175,1,137,6,233,199,243,82,177,15,236,8,5],\"binding_number\":[214,221,36,34,246,89,38,85,207,68,157,111,72,61,241,253,70,30,206,120,80,185,10,217,210,162,9,77,57,112,79,112],\"first_attribute\":[213,53,84,155,222,190,233,7,127,153,97,181,71,169,88,180,33,255,144,158,221,62,119,61,11,183,140,136,198,152,186,21],\"second_attribute\":[231,35,35,8,13,210,132,121,142,177,211,196,174,183,10,161,8,232,101,123,107,10,144,104,58,212,95,243,216,98,91,66],\"blind_sign_req\":[131,90,160,125,41,86,193,108,153,195,118,191,72,114,19,168,89,70,211,33,230,221,202,233,172,194,91,92,168,79,223,65,217,200,186,142,16,136,216,83,40,22,53,249,50,29,234,219,174,135,50,180,67,16,30,73,89,232,51,155,170,105,83,240,116,155,50,214,28,98,241,211,27,202,201,186,231,151,9,249,149,50,246,24,204,81,141,77,51,32,90,254,63,150,35,228,2,0,0,0,0,0,0,0,141,71,143,23,11,175,34,130,24,40,235,202,85,74,58,208,132,184,176,43,159,38,159,138,118,16,141,70,20,77,59,12,153,184,153,255,181,243,192,131,239,253,207,58,228,82,207,244,143,126,159,156,127,116,223,200,175,191,130,91,218,19,169,222,95,36,151,19,18,14,236,135,21,30,69,130,146,236,252,85,152,68,107,154,102,48,200,246,88,205,96,74,12,203,10,2,88,70,108,249,72,221,172,76,212,26,105,127,1,214,165,143,133,241,142,106,52,190,152,116,101,26,181,213,208,59,117,33,128,70,252,146,203,142,173,98,252,135,68,44,197,243,225,52,170,142,92,167,250,23,211,161,7,85,102,108,15,164,238,94,2,0,0,0,0,0,0,0,4,48,16,6,220,119,137,9,19,83,33,21,156,10,94,247,114,65,199,182,184,76,83,122,118,46,29,9,111,242,244,4,195,174,63,91,223,161,105,250,195,94,75,226,206,80,226,190,37,203,200,184,173,164,170,226,27,244,43,100,95,65,40,45,2,0,0,0,0,0,0,0,136,46,57,4,42,1,115,237,142,99,92,81,193,81,203,189,226,255,137,255,109,205,71,147,74,191,253,120,93,154,61,18,139,163,156,182,203,137,253,197,11,225,221,81,59,38,23,43,168,53,214,23,140,227,39,174,93,191,83,150,228,30,41,12]},\"signature\":\"237zXmYcb2de47menhqKVUq75YUQYmDfcZ2Cnpqf7c69i1dpK4DetfzxUmsZxkoEw3pfBZkbkZdAUHbXpKJRaVG9RVy13gucaiZVpnmrdHmyqsfUJYqxB5XJy7PvnQufvW1p\"}","AB1DB6BCE0E5BB8093CD83CE8B979A38ABA549017F1E5FBCFC6665D8301F8B10":"{\"amount\":1000000,\"tx_hash\":\"AB1DB6BCE0E5BB8093CD83CE8B979A38ABA549017F1E5FBCFC6665D8301F8B10\",\"signing_keypair\":{\"public_key\":\"BsW9Zi5fr6MVXVD9C79DHazZngKYVU29Dd2FqiB3t6GC\",\"private_key\":\"5t6asn1vRZunc4tna9doZoURJJyVsugmkDwgaQe6fmZg\"},\"encryption_keypair\":{\"public_key\":\"Gx3p6FJxBs7uEAQPWFmVwiysPBeRseC5XvUixiNFFeAt\",\"private_key\":\"CKV5eAdhsmyBijmnt1UM7wLTmxnLxbV6h2WY6JFZmYVm\"},\"blind_request_data\":{\"serial_number\":[98,22,102,53,67,142,112,200,210,202,159,104,167,22,128,165,252,131,183,212,4,150,175,203,141,146,137,193,136,219,228,67],\"binding_number\":[3,214,193,92,3,51,114,82,60,25,195,249,143,244,252,169,19,84,168,119,227,141,186,35,187,120,96,205,72,61,189,48],\"first_attribute\":[132,126,154,107,2,30,197,238,4,205,66,148,113,208,48,233,37,53,187,160,213,151,49,171,64,57,139,90,63,87,188,115],\"second_attribute\":[106,78,124,199,142,38,90,198,100,219,16,74,9,109,47,212,166,30,185,234,204,112,42,38,193,74,135,219,223,179,30,19],\"blind_sign_req\":[145,90,96,73,107,35,232,107,92,90,101,110,229,20,12,197,185,112,19,110,4,212,236,222,33,252,156,12,164,253,198,80,41,254,246,26,247,150,242,56,2,250,155,142,20,132,108,172,178,146,141,193,156,246,79,110,32,254,166,217,57,83,8,70,58,42,48,48,137,96,13,193,214,125,89,122,85,97,13,112,29,193,119,165,251,204,221,81,138,206,108,3,4,88,182,27,2,0,0,0,0,0,0,0,131,68,160,172,144,26,48,222,7,103,95,220,238,3,125,119,15,254,236,186,107,240,69,183,43,182,191,139,208,201,100,128,55,146,133,119,213,155,40,224,106,149,217,30,147,95,52,7,146,147,130,119,30,74,157,239,231,11,199,90,188,64,201,48,27,156,69,63,33,107,61,254,128,200,127,207,97,179,198,142,223,95,167,197,54,54,54,28,122,22,99,197,213,105,56,211,204,241,133,3,184,107,81,82,10,67,188,3,221,93,229,232,125,173,124,242,248,183,14,124,52,120,234,79,175,133,191,33,45,79,152,252,100,22,220,193,244,169,43,29,202,104,2,96,164,95,116,46,5,23,50,139,39,23,73,151,94,229,153,21,2,0,0,0,0,0,0,0,90,175,157,204,225,13,7,24,166,246,95,121,101,58,127,194,67,222,14,175,13,126,221,162,91,86,130,91,249,183,86,48,164,151,67,190,88,250,83,103,236,57,114,53,155,50,92,181,58,239,141,200,153,16,44,83,110,125,190,133,195,145,165,98,2,0,0,0,0,0,0,0,45,195,137,247,231,86,230,71,21,193,215,245,38,74,249,49,81,90,142,237,9,123,34,252,140,107,120,2,118,147,121,107,65,0,99,209,123,64,124,244,142,136,130,186,77,32,148,12,216,71,76,245,179,241,249,177,44,191,107,95,159,123,125,52]},\"signature\":\"24WoBrFcv1mF8fL2MH1YHnbZq69G44ZqSzbdA8ciEWY8JXTFn1G6yyKGEUkxcmLoL7r5j9fPJD4mTZUdiRQ87pLMZPCkp6M9bLZMwD7b4u2Cg8CNmVUpaZ4VbNdTf4zhZid5\"}","90B33BF08C84575E88C6ACFFE6F44DFC652BF107CDC2D541BFCA8BAC931E5BA8":"{\"amount\":1000000,\"tx_hash\":\"90B33BF08C84575E88C6ACFFE6F44DFC652BF107CDC2D541BFCA8BAC931E5BA8\",\"signing_keypair\":{\"public_key\":\"4o64HhsgHq6bCWERouYe8wLGdrBW6Zc87uRFsRyQ6KhX\",\"private_key\":\"9pcuX5cpKxGde6ZiPbsMbSKBx8qcnbSD5VkV68VuNQB3\"},\"encryption_keypair\":{\"public_key\":\"2cF4745DbeYYxymmPJYTu9R88wRVSHnC7rs4TPvLkNxm\",\"private_key\":\"6wyfg3FEwbtWJ4pAfwEVvVuo4gEJV6cc4Kj6LV3qy2WV\"},\"blind_request_data\":{\"serial_number\":[95,129,65,207,163,115,235,106,18,228,138,56,255,52,168,5,190,226,124,190,211,158,70,124,38,235,132,6,25,222,157,31],\"binding_number\":[147,54,129,176,80,3,106,255,93,179,219,233,158,175,141,145,180,245,103,169,36,3,99,138,7,139,108,186,186,193,152,17],\"first_attribute\":[61,240,7,131,88,85,251,196,132,143,227,154,197,40,194,163,27,24,53,182,97,31,61,238,241,172,115,236,99,2,240,15],\"second_attribute\":[203,202,64,166,201,186,160,157,155,156,127,116,239,44,158,213,167,57,152,228,39,153,194,20,46,222,44,246,87,151,164,61],\"blind_sign_req\":[146,176,228,40,252,11,192,170,3,84,188,29,87,64,49,134,204,118,180,111,166,23,44,91,115,7,5,41,143,230,192,139,183,60,135,118,68,33,28,56,245,7,61,128,246,183,224,155,149,180,159,139,44,6,86,185,32,243,35,50,208,167,105,51,10,199,17,25,202,242,82,185,180,93,181,71,9,21,198,3,53,27,228,247,70,247,15,231,35,239,171,168,12,39,9,103,2,0,0,0,0,0,0,0,149,10,49,53,253,27,20,83,77,16,209,183,26,202,173,245,48,2,98,214,50,111,205,61,143,205,180,12,60,80,204,8,225,176,250,242,51,147,6,172,49,20,234,128,58,157,77,9,173,73,60,198,152,161,17,123,43,108,134,157,145,182,38,23,7,244,179,16,244,222,206,4,241,95,210,165,18,216,116,249,134,68,231,240,47,49,30,18,236,98,51,218,81,213,208,120,166,39,74,65,151,62,195,247,123,40,224,232,120,182,234,214,63,185,102,225,19,65,65,28,233,91,168,201,249,193,43,59,2,192,145,29,106,92,124,57,210,121,76,30,81,115,226,84,155,68,97,46,82,34,160,51,76,110,123,248,55,124,137,102,2,0,0,0,0,0,0,0,237,8,254,97,90,184,163,156,221,82,133,94,177,40,166,132,230,14,241,234,176,194,121,0,157,148,70,12,157,15,160,28,212,94,219,169,220,163,101,233,97,95,106,244,208,115,209,2,3,150,129,201,126,49,98,125,26,68,122,103,142,111,23,21,2,0,0,0,0,0,0,0,93,65,229,251,193,25,208,246,197,112,140,170,138,227,58,92,23,231,190,23,95,49,20,247,18,65,168,78,40,181,74,69,7,14,136,243,70,224,141,79,202,202,127,255,64,28,8,79,69,17,47,130,129,224,103,10,50,89,97,99,93,242,119,14]},\"signature\":\"ta3pM9ffj5T6YGbwjSBp2W118rcwyP9PXStc7ssb91g5GQYMQHhuTNajbdZcjxUFBFL5rhED8EHpRzE8r432ss3qbPBfpNev4CdkfMkQ3wepyM7hy7q1W6Rn9WmFoZLZR9j\"}"},{}] \ No newline at end of file diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index f4e411652c..2c3dc578f2 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -16,7 +16,7 @@ use proxy_helpers::connection_controller::{ }; use proxy_helpers::proxy_runner::ProxyRunner; use rand::RngCore; -use socks5_requests::{ConnectionId, RemoteAddress, Request}; +use socks5_requests::{ConnectionId, Message, RemoteAddress, Request}; use std::io; use std::net::SocketAddr; use std::pin::Pin; @@ -224,8 +224,9 @@ impl SocksClient { async fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) { let req = Request::new_connect(self.connection_id, remote_address, self.self_address); + let msg = Message::Request(req); - let input_message = InputMessage::new_fresh(self.service_provider, req.into_bytes(), false); + let input_message = InputMessage::new_fresh(self.service_provider, msg.into_bytes(), false); self.input_sender.unbounded_send(input_message).unwrap(); } @@ -252,7 +253,8 @@ impl SocksClient { ) .run(move |conn_id, read_data, socket_closed| { let provider_request = Request::new_send(conn_id, read_data, socket_closed); - InputMessage::new_fresh(recipient, provider_request.into_bytes(), false) + let provider_message = Message::Request(provider_request); + InputMessage::new_fresh(recipient, provider_message.into_bytes(), false) }) .await .into_inner(); diff --git a/common/socks5/requests/src/lib.rs b/common/socks5/requests/src/lib.rs index 6415885625..434bf0df24 100644 --- a/common/socks5/requests/src/lib.rs +++ b/common/socks5/requests/src/lib.rs @@ -1,5 +1,7 @@ +pub mod msg; pub mod request; pub mod response; +pub use msg::*; pub use request::*; pub use response::*; diff --git a/common/socks5/requests/src/msg.rs b/common/socks5/requests/src/msg.rs new file mode 100644 index 0000000000..21a37b1190 --- /dev/null +++ b/common/socks5/requests/src/msg.rs @@ -0,0 +1,63 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::request::{Request, RequestError}; +use crate::response::{Response, ResponseError}; + +#[derive(Debug)] +pub enum MessageError { + Request(RequestError), + Response(ResponseError), + NoData, + UnknownMessageType, +} + +impl std::fmt::Display for MessageError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + MessageError::Request(r) => write!(f, "{}", r), + MessageError::Response(r) => write!(f, "{:?}", r), + MessageError::NoData => write!(f, "no data provided"), + MessageError::UnknownMessageType => write!(f, "unknown message type received"), + } + } +} + +pub enum Message { + Request(Request), + Response(Response), +} + +impl Message { + const REQUEST_FLAG: u8 = 0; + const RESPONSE_FLAG: u8 = 1; + + pub fn try_from_bytes(b: &[u8]) -> Result { + if b.is_empty() { + return Err(MessageError::NoData); + } + + if b[0] == Self::REQUEST_FLAG { + Request::try_from_bytes(&b[1..]) + .map(Message::Request) + .map_err(MessageError::Request) + } else if b[0] == Self::RESPONSE_FLAG { + Response::try_from_bytes(&b[1..]) + .map(Message::Response) + .map_err(MessageError::Response) + } else { + Err(MessageError::UnknownMessageType) + } + } + + pub fn into_bytes(self) -> Vec { + match self { + Self::Request(r) => std::iter::once(Self::REQUEST_FLAG) + .chain(r.into_bytes().iter().cloned()) + .collect(), + Self::Response(r) => std::iter::once(Self::RESPONSE_FLAG) + .chain(r.into_bytes().iter().cloned()) + .collect(), + } + } +} diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index a33bdb0bee..1be27ee64d 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -10,23 +10,36 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +bincode = "1.3" clap = "2.33.0" dirs = "3.0" futures = "0.3" +ipnetwork = "0.17" log = "0.4" pretty_env_logger = "0.4" -tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros" ] } -tokio-tungstenite = "0.14" publicsuffix = "1.5" -ipnetwork = "0.17" +rocket = { version = "0.5.0-rc.1", features = ["json"], optional = true } +serde = { version = "1.0", features = ["derive"] } +sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]} +thiserror = "1" +tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros", "time" ] } +tokio-tungstenite = "0.14" # internal +network-defaults = { path = "../../common/network-defaults" } nymsphinx = { path = "../../common/nymsphinx" } ordered-buffer = {path = "../../common/socks5/ordered-buffer"} -socks5-requests = { path = "../../common/socks5/requests" } proxy-helpers = { path = "../../common/socks5/proxy-helpers" } +socks5-requests = { path = "../../common/socks5/requests" } websocket-requests = { path = "../../clients/native/websocket-requests" } [dev-dependencies] rand = "0.7" + +[build-dependencies] +sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } +tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] } + +[features] +stats-service = ["rocket"] \ No newline at end of file diff --git a/service-providers/network-requester/README.md b/service-providers/network-requester/README.md new file mode 100644 index 0000000000..a5006561ef --- /dev/null +++ b/service-providers/network-requester/README.md @@ -0,0 +1,26 @@ +Copyright 2022 - Nym Technologies SA +SPDX-License-Identifier: Apache-2.0 +--> + +## Network requester + +The network requester is used to interpret socks5 client messages that need to +be proxied to a running service i.e. a host and a port. + +If you have a service that you want to expose to the mixnet, you'd need to +first run the native client and provide the client address to your users that +will use it in their socks5 configuration. + +After starting the native client, start the network requester and configure it, +setting your service's endpoint in +`${HOME}/.nym/service-providers/network-requester/allowed.list` + +Running in `open-proxy` mode allows any traffic to be proxied by the network +requester. + +### Statistics service +The network requester can be build and ran as a gatherer of statistics from all +the other network requesters on the mixnet. For that, build the binary with the +`stats-service` feature enabled. The native client address that corresponds to +this network requester would have to be built into the constants of all the +other network requesters that are sending the data. \ No newline at end of file diff --git a/service-providers/network-requester/build.rs b/service-providers/network-requester/build.rs new file mode 100644 index 0000000000..73e5fdf775 --- /dev/null +++ b/service-providers/network-requester/build.rs @@ -0,0 +1,28 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use sqlx::{Connection, SqliteConnection}; +use std::env; + +#[tokio::main] +async fn main() { + let out_dir = env::var("OUT_DIR").unwrap(); + let database_path = format!("{}/network-requester-example.sqlite", out_dir); + + let mut conn = SqliteConnection::connect(&*format!("sqlite://{}?mode=rwc", database_path)) + .await + .expect("Failed to create SQLx database connection"); + + sqlx::migrate!("./migrations") + .run(&mut conn) + .await + .expect("Failed to perform SQLx migrations"); + + #[cfg(target_family = "unix")] + println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path); + + #[cfg(target_family = "windows")] + // for some strange reason we need to add a leading `/` to the windows path even though it's + // not a valid windows path... but hey, it works... + println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path); +} diff --git a/service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql b/service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql new file mode 100644 index 0000000000..c051557da2 --- /dev/null +++ b/service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql @@ -0,0 +1,14 @@ +/* + * Copyright 2022 - Nym Technologies SA + * SPDX-License-Identifier: Apache-2.0 + */ + +CREATE TABLE mixnet_statistics +( + id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, + service_description VARCHAR NOT NULL, + request_processed_bytes INTEGER NOT NULL, + response_processed_bytes INTEGER NOT NULL, + interval_seconds INTEGER NOT NULL, + timestamp DATETIME NOT NULL +); \ No newline at end of file diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 43b317b3ee..ee5dbc4394 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -3,6 +3,7 @@ use crate::allowed_hosts::{HostsStore, OutboundRequestFilter}; use crate::connection::Connection; +use crate::statistics::{Statistics, StatsData, Timer}; use crate::websocket; use crate::websocket::TSWebsocketStream; use futures::channel::mpsc; @@ -12,9 +13,11 @@ use log::*; use nymsphinx::addressing::clients::Recipient; use nymsphinx::receiver::ReconstructedMessage; use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender}; -use socks5_requests::{ConnectionId, Request, Response}; +use socks5_requests::{ConnectionId, Message as Socks5Message, Request, Response}; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use tokio::sync::RwLock; use tokio_tungstenite::tungstenite::protocol::Message; use websocket::WebsocketConnectionError; use websocket_requests::{requests::ClientRequest, responses::ServerResponse}; @@ -24,12 +27,19 @@ static ACTIVE_PROXIES: AtomicUsize = AtomicUsize::new(0); pub struct ServiceProvider { listening_address: String, + description: String, + #[cfg(feature = "stats-service")] + db_path: PathBuf, outbound_request_filter: OutboundRequestFilter, open_proxy: bool, } impl ServiceProvider { - pub fn new(listening_address: String, open_proxy: bool) -> ServiceProvider { + pub fn new( + listening_address: String, + description: String, + open_proxy: bool, + ) -> ServiceProvider { let allowed_hosts = HostsStore::new( HostsStore::default_base_dir(), PathBuf::from("allowed.list"), @@ -39,9 +49,19 @@ impl ServiceProvider { HostsStore::default_base_dir(), PathBuf::from("unknown.list"), ); + + #[cfg(feature = "stats-service")] + let db_path = HostsStore::default_base_dir() + .join("service-providers") + .join("network-requester") + .join("db.sqlite"); + let outbound_request_filter = OutboundRequestFilter::new(allowed_hosts, unknown_hosts); ServiceProvider { listening_address, + description, + #[cfg(feature = "stats-service")] + db_path, outbound_request_filter, open_proxy, } @@ -52,13 +72,18 @@ impl ServiceProvider { async fn mixnet_response_listener( mut websocket_writer: SplitSink, mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>, + response_stats_data: &Arc>, ) { // TODO: wire SURBs in here once they're available while let Some((response, return_address)) = mix_reader.next().await { + response_stats_data + .write() + .await + .processed(response.data.len() as u32); // make 'request' to native-websocket client let response_message = ClientRequest::Send { recipient: return_address, - message: response.into_bytes(), + message: Socks5Message::Response(response).into_bytes(), with_reply_surb: false, }; @@ -194,31 +219,50 @@ impl ServiceProvider { .unwrap() } - fn handle_proxy_request( + async fn handle_proxy_message( &mut self, + #[cfg(feature = "stats-service")] storage: &crate::storage::NetworkRequesterStorage, raw_request: &[u8], controller_sender: &mut ControllerSender, mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>, + request_stats_data: &Arc>, ) { - // try to treat each received mix message as a service provider request - let deserialized_request = match Request::try_from_bytes(raw_request) { - Ok(request) => request, + let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) { + Ok(msg) => msg, Err(err) => { - error!("Failed to deserialized received request! - {}", err); + error!("Failed to deserialized received message! - {}", err); return; } }; + match deserialized_msg { + Socks5Message::Request(deserialized_request) => match deserialized_request { + Request::Connect(req) => self.handle_proxy_connect( + controller_sender, + mix_input_sender, + req.conn_id, + req.remote_addr, + req.return_address, + ), - match deserialized_request { - Request::Connect(req) => self.handle_proxy_connect( - controller_sender, - mix_input_sender, - req.conn_id, - req.remote_addr, - req.return_address, - ), - Request::Send(conn_id, data, closed) => { - self.handle_proxy_send(controller_sender, conn_id, data, closed) + Request::Send(conn_id, data, closed) => { + request_stats_data + .write() + .await + .processed(data.len() as u32); + self.handle_proxy_send(controller_sender, conn_id, data, closed) + } + }, + Socks5Message::Response(_deserialized_response) => + { + #[cfg(feature = "stats-service")] + match crate::statistics::StatsMessage::from_bytes(&_deserialized_response.data) { + Ok(data) => { + if let Err(e) = storage.insert_service_statistics(data).await { + error!("Could not store received statistics: {}", e); + } + } + Err(e) => error!("Malformed statistics received: {}", e), + } } } } @@ -234,19 +278,56 @@ impl ServiceProvider { // going to be used by `mixnet_response_listener` let (mix_input_sender, mix_input_receiver) = mpsc::unbounded::<(Response, Recipient)>(); + let (mut timer_sender, timer_receiver) = Timer::new(); + let interval = timer_sender.interval(); + tokio::spawn(async move { + timer_sender.run().await; + }); + // controller for managing all active connections let (mut active_connections_controller, mut controller_sender) = Controller::new(); tokio::spawn(async move { active_connections_controller.run().await; }); + let mut stats = Statistics::new(self.description.clone(), interval, timer_receiver); + let request_stats_data = Arc::clone(stats.request_data()); + let response_stats_data = Arc::clone(stats.response_data()); + let mix_input_sender_clone = mix_input_sender.clone(); + tokio::spawn(async move { + stats.run(&mix_input_sender_clone).await; + }); + + #[cfg(feature = "stats-service")] + let storage = crate::storage::NetworkRequesterStorage::init(&self.db_path) + .await + .expect("Could not create network requester storage"); + + #[cfg(feature = "stats-service")] + tokio::spawn( + rocket::build() + .mount( + "/v1", + rocket::routes![crate::storage::post_mixnet_statistics], + ) + .manage(storage.clone()) + .ignite() + .await + .expect("Could not ignite stats api service") + .launch(), + ); + // start the listener for mix messages tokio::spawn(async move { - Self::mixnet_response_listener(websocket_writer, mix_input_receiver).await; + Self::mixnet_response_listener( + websocket_writer, + mix_input_receiver, + &response_stats_data, + ) + .await; }); println!("\nAll systems go. Press CTRL-C to stop the server."); - // for each incoming message from the websocket... (which in 99.99% cases is going to be a mix message) loop { let received = match Self::read_websocket_message(&mut websocket_reader).await { @@ -260,7 +341,15 @@ impl ServiceProvider { let raw_message = received.message; // TODO: here be potential SURB (i.e. received.reply_SURB) - self.handle_proxy_request(&raw_message, &mut controller_sender, &mix_input_sender) + self.handle_proxy_message( + #[cfg(feature = "stats-service")] + &storage, + &raw_message, + &mut controller_sender, + &mix_input_sender, + &request_stats_data, + ) + .await; } } diff --git a/service-providers/network-requester/src/main.rs b/service-providers/network-requester/src/main.rs index 1780521a53..935f37b6c1 100644 --- a/service-providers/network-requester/src/main.rs +++ b/service-providers/network-requester/src/main.rs @@ -3,12 +3,19 @@ use clap::{App, Arg, ArgMatches}; +use network_defaults::DEFAULT_WEBSOCKET_LISTENING_PORT; + mod allowed_hosts; mod connection; mod core; +mod statistics; +#[cfg(feature = "stats-service")] +mod storage; mod websocket; const OPEN_PROXY_ARG: &str = "open-proxy"; +const WS_PORT: &str = "websocket-port"; +const DESCRIPTION: &str = "description"; fn parse_args<'a>() -> ArgMatches<'a> { App::new("Nym Network Requester") @@ -20,6 +27,21 @@ fn parse_args<'a>() -> ArgMatches<'a> { .long(OPEN_PROXY_ARG) .short("o"), ) + .arg( + Arg::with_name(WS_PORT) + .help("websocket port to bind to") + .long(WS_PORT) + .short("p") + .takes_value(true), + ) + .arg( + Arg::with_name(DESCRIPTION) + .help("service description") + .long(DESCRIPTION) + .short("d") + .takes_value(true) + .default_value("undefined"), + ) .get_matches() } @@ -27,14 +49,22 @@ fn parse_args<'a>() -> ArgMatches<'a> { async fn main() { setup_logging(); let matches = parse_args(); + let open_proxy = matches.is_present(OPEN_PROXY_ARG); if open_proxy { println!("\n\nYOU HAVE STARTED IN 'OPEN PROXY' MODE. ANYONE WITH YOUR CLIENT ADDRESS CAN MAKE REQUESTS FROM YOUR MACHINE. PLEASE QUIT IF YOU DON'T UNDERSTAND WHAT YOU'RE DOING.\n\n"); } - let uri = "ws://localhost:1977"; + let uri = format!( + "ws://localhost:{}", + matches + .value_of(WS_PORT) + .unwrap_or(&DEFAULT_WEBSOCKET_LISTENING_PORT.to_string()) + ); + + let description = matches.value_of(DESCRIPTION).unwrap().to_string(); println!("Starting socks5 service provider:"); - let mut server = core::ServiceProvider::new(uri.into(), open_proxy); + let mut server = core::ServiceProvider::new(uri, description, open_proxy); server.run().await; } diff --git a/service-providers/network-requester/src/statistics/comm.rs b/service-providers/network-requester/src/statistics/comm.rs new file mode 100644 index 0000000000..4ba1d7e62a --- /dev/null +++ b/service-providers/network-requester/src/statistics/comm.rs @@ -0,0 +1,139 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use futures::channel::mpsc; +use futures::StreamExt; +use log::*; +use serde::{Deserialize, Serialize}; +use sqlx::types::chrono::{DateTime, Utc}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::RwLock; + +use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipient}; +use nymsphinx::addressing::nodes::NodeIdentity; +use socks5_requests::Response; + +use super::error::StatsError; + +const STATS_PROVIDER_CLIENT_IDENTITY: &str = "HqYWvCcB4sswYiyMj5Q8H5oc71kLf96vfrLK3npM7stH"; +const STATS_PROVIDER_ENCRYPTION_KEY: &str = "CoeC5dcqurgdxr5zcgU77nZBSBCc8ntCiwUivQ9TX3KT"; +const STATS_PROVIDER_GATEWAY_IDENTITY: &str = "E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM"; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct StatsMessage { + pub description: String, + pub request_data: StatsData, + pub response_data: StatsData, + pub interval_seconds: u32, + pub timestamp: String, +} + +impl StatsMessage { + pub fn to_bytes(&self) -> Result, StatsError> { + Ok(bincode::serialize(self)?) + } + + #[cfg(feature = "stats-service")] + pub fn from_bytes(b: &[u8]) -> Result { + Ok(bincode::deserialize(b)?) + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct StatsData { + total_processed_bytes: u32, +} + +impl StatsData { + pub fn new(total_processed_bytes: u32) -> Self { + StatsData { + total_processed_bytes, + } + } + pub fn processed(&mut self, bytes: u32) { + self.total_processed_bytes += bytes; + } + + #[cfg(feature = "stats-service")] + pub fn total_processed_bytes(&self) -> u32 { + self.total_processed_bytes + } +} + +pub struct Statistics { + description: String, + request_data: Arc>, + response_data: Arc>, + interval_seconds: u32, + timestamp: DateTime, + timer_receiver: mpsc::Receiver<()>, + stats_provider_addr: Recipient, +} + +impl Statistics { + pub fn new( + description: String, + interval_seconds: Duration, + timer_receiver: mpsc::Receiver<()>, + ) -> Self { + // those unwraps are ok because we set the strings in the constants above + let stats_provider_addr = Recipient::new( + ClientIdentity::from_base58_string(STATS_PROVIDER_CLIENT_IDENTITY).unwrap(), + ClientEncryptionKey::from_base58_string(STATS_PROVIDER_ENCRYPTION_KEY).unwrap(), + NodeIdentity::from_base58_string(STATS_PROVIDER_GATEWAY_IDENTITY).unwrap(), + ); + Statistics { + description, + request_data: Arc::new(RwLock::new(StatsData::new(0))), + response_data: Arc::new(RwLock::new(StatsData::new(0))), + timestamp: Utc::now(), + interval_seconds: interval_seconds.as_secs() as u32, + timer_receiver, + stats_provider_addr, + } + } + + pub fn request_data(&self) -> &Arc> { + &self.request_data + } + + pub fn response_data(&self) -> &Arc> { + &self.response_data + } + + pub async fn run(&mut self, mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>) { + loop { + if self.timer_receiver.next().await == None { + error!("Timer thread has died. No more statistics will be sent"); + } else { + let stats_message = StatsMessage { + description: self.description.clone(), + request_data: self.request_data.read().await.clone(), + response_data: self.response_data.read().await.clone(), + interval_seconds: self.interval_seconds, + timestamp: self.timestamp.to_rfc3339(), + }; + match stats_message.to_bytes() { + Ok(data) => { + trace!("Sending data to statistics service"); + mix_input_sender + .unbounded_send(( + Response::new(0, data, false), + self.stats_provider_addr, + )) + .unwrap(); + } + Err(e) => error!("Statistics not sent: {}", e), + } + self.reset_stats().await; + } + } + } + + async fn reset_stats(&mut self) { + self.request_data.write().await.total_processed_bytes = 0; + self.response_data.write().await.total_processed_bytes = 0; + self.timestamp = Utc::now(); + } +} diff --git a/service-providers/network-requester/src/statistics/error.rs b/service-providers/network-requester/src/statistics/error.rs new file mode 100644 index 0000000000..f29b79b91d --- /dev/null +++ b/service-providers/network-requester/src/statistics/error.rs @@ -0,0 +1,10 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum StatsError { + #[error("Bincode error: {0}")] + BincodeError(#[from] bincode::Error), +} diff --git a/service-providers/network-requester/src/statistics/mod.rs b/service-providers/network-requester/src/statistics/mod.rs new file mode 100644 index 0000000000..6ba821cf5f --- /dev/null +++ b/service-providers/network-requester/src/statistics/mod.rs @@ -0,0 +1,9 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +mod comm; +mod error; +mod timer; + +pub use comm::{Statistics, StatsData, StatsMessage}; +pub use timer::Timer; diff --git a/service-providers/network-requester/src/statistics/timer.rs b/service-providers/network-requester/src/statistics/timer.rs new file mode 100644 index 0000000000..cde054c931 --- /dev/null +++ b/service-providers/network-requester/src/statistics/timer.rs @@ -0,0 +1,40 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use futures::channel::mpsc; +use std::time::Duration; +use tokio::time; + +const STATISTICS_INTERVAL: Duration = Duration::from_secs(60); + +pub type TimerReceiver = mpsc::Receiver<()>; + +pub struct Timer { + interval: Duration, + stats_sender: mpsc::Sender<()>, +} + +impl Timer { + pub fn new() -> (Self, TimerReceiver) { + let (stats_sender, stats_receiver) = mpsc::channel::<()>(1); + ( + Timer { + interval: STATISTICS_INTERVAL, + stats_sender, + }, + stats_receiver, + ) + } + + pub fn interval(&self) -> Duration { + self.interval + } + + pub async fn run(&mut self) { + let mut interval = time::interval(self.interval); + loop { + interval.tick().await; + let _ = self.stats_sender.try_send(()); + } + } +} diff --git a/service-providers/network-requester/src/storage/error.rs b/service-providers/network-requester/src/storage/error.rs new file mode 100644 index 0000000000..1df83a46fd --- /dev/null +++ b/service-providers/network-requester/src/storage/error.rs @@ -0,0 +1,14 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +#[derive(Debug, thiserror::Error)] +pub enum NetworkRequesterStorageError { + #[error("SQL error - {0}")] + InternalDatabaseError(#[from] sqlx::Error), + + #[error("SQL migrate error - {0}")] + DatabaseMigrateError(#[from] sqlx::migrate::MigrateError), + + #[error("Timestamp could not be parsed")] + TimestampParse, +} diff --git a/service-providers/network-requester/src/storage/manager.rs b/service-providers/network-requester/src/storage/manager.rs new file mode 100644 index 0000000000..1d91f65829 --- /dev/null +++ b/service-providers/network-requester/src/storage/manager.rs @@ -0,0 +1,62 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use sqlx::types::chrono::{DateTime, Utc}; + +use crate::storage::models::MixnetStatistics; + +#[derive(Clone)] +pub(crate) struct StorageManager { + pub(crate) connection_pool: sqlx::SqlitePool, +} + +// all SQL goes here +impl StorageManager { + /// Adds an entry for some statistical data. + /// + /// # Arguments + /// + /// * `service_description`: Description of the service that gathered the data. + /// * `request_processed_bytes`: Number of bytes for socks5 requests. + /// * `response_processed_bytes`: Number of bytes for socks5 responses. + /// * `interval_seconds`: Duration in seconds in which the data was gathered. + pub(super) async fn insert_service_statistics( + &self, + service_description: String, + request_processed_bytes: u32, + response_processed_bytes: u32, + interval_seconds: u32, + timestamp: DateTime, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "INSERT INTO mixnet_statistics(service_description, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?)", + service_description, + request_processed_bytes, + response_processed_bytes, + interval_seconds, + timestamp, + ).execute(&self.connection_pool).await?; + Ok(()) + } + + /// Returns data submitted within the provided time interval. + /// + /// # Arguments + /// + /// * `since`: indicates the lower bound timestamp for the data + /// * `until`: indicates the upper bound timestamp for the data + pub(super) async fn get_service_statistics_in_interval( + &self, + since: DateTime, + until: DateTime, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + MixnetStatistics, + "SELECT * FROM mixnet_statistics WHERE timestamp BETWEEN ? AND ?", + since, + until + ) + .fetch_all(&self.connection_pool) + .await + } +} diff --git a/service-providers/network-requester/src/storage/mod.rs b/service-providers/network-requester/src/storage/mod.rs new file mode 100644 index 0000000000..1cf74a7778 --- /dev/null +++ b/service-providers/network-requester/src/storage/mod.rs @@ -0,0 +1,92 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use log::*; +use sqlx::types::chrono::{DateTime, Utc}; +use sqlx::ConnectOptions; +use std::path::PathBuf; + +use crate::statistics::StatsMessage; +use crate::storage::error::NetworkRequesterStorageError; +use crate::storage::manager::StorageManager; +use crate::storage::models::MixnetStatistics; +pub(crate) use crate::storage::routes::post_mixnet_statistics; + +mod error; +mod manager; +mod models; +mod routes; + +// note that clone here is fine as upon cloning the same underlying pool will be used +#[derive(Clone)] +pub(crate) struct NetworkRequesterStorage { + manager: StorageManager, +} + +impl NetworkRequesterStorage { + pub async fn init(database_path: &PathBuf) -> Result { + let mut opts = sqlx::sqlite::SqliteConnectOptions::new() + .filename(database_path) + .create_if_missing(true); + + opts.disable_statement_logging(); + + let connection_pool = sqlx::SqlitePool::connect_with(opts).await?; + + sqlx::migrate!("./migrations").run(&connection_pool).await?; + info!("Database migration finished!"); + + let storage = NetworkRequesterStorage { + manager: StorageManager { connection_pool }, + }; + + Ok(storage) + } + + /// Adds an entry for some statistical data. + /// + /// # Arguments + /// + /// * `msg`: Message containing the statistical data. + pub(super) async fn insert_service_statistics( + &self, + msg: StatsMessage, + ) -> Result<(), NetworkRequesterStorageError> { + let timestamp: DateTime = DateTime::parse_from_rfc3339(&msg.timestamp) + .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .into(); + Ok(self + .manager + .insert_service_statistics( + msg.description, + msg.request_data.total_processed_bytes(), + msg.response_data.total_processed_bytes(), + msg.interval_seconds, + timestamp, + ) + .await?) + } + + /// Returns data submitted within the provided time interval. + /// + /// # Arguments + /// + /// * `since`: indicates the lower bound timestamp for the data, RFC 3339 format + /// * `until`: indicates the upper bound timestamp for the data, RFC 3339 format + pub(super) async fn get_service_statistics_in_interval( + &self, + since: &str, + until: &str, + ) -> Result, NetworkRequesterStorageError> { + let since = DateTime::parse_from_rfc3339(since) + .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .into(); + let until = DateTime::parse_from_rfc3339(until) + .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .into(); + Ok(self + .manager + .get_service_statistics_in_interval(since, until) + .await?) + } +} diff --git a/service-providers/network-requester/src/storage/models.rs b/service-providers/network-requester/src/storage/models.rs new file mode 100644 index 0000000000..e160fe703e --- /dev/null +++ b/service-providers/network-requester/src/storage/models.rs @@ -0,0 +1,15 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use sqlx::types::chrono::NaiveDateTime; + +// Internally used struct to catch results from the database to get mixnet statistics +pub(crate) struct MixnetStatistics { + #[allow(dead_code)] + pub(crate) id: i64, + pub(crate) service_description: String, + pub(crate) request_processed_bytes: i64, + pub(crate) response_processed_bytes: i64, + pub(crate) interval_seconds: i64, + pub(crate) timestamp: NaiveDateTime, +} diff --git a/service-providers/network-requester/src/storage/routes.rs b/service-providers/network-requester/src/storage/routes.rs new file mode 100644 index 0000000000..638317048d --- /dev/null +++ b/service-providers/network-requester/src/storage/routes.rs @@ -0,0 +1,49 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use rocket::serde::json::Json; +use rocket::State; +use serde::{Deserialize, Serialize}; + +use crate::statistics::{StatsData, StatsMessage}; +use crate::storage::NetworkRequesterStorage; + +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct MixnetStatisticsRequest { + // date, RFC 3339 format + since: String, + // date, RFC 3339 format + until: String, +} + +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct MixnetStatisticsResponse { + data: Vec, +} + +#[rocket::post("/mixnet-statistics", data = "")] +pub(crate) async fn post_mixnet_statistics( + mixnet_statistics_request: Json, + storage: &State, +) -> Json { + let mixnet_statistics = storage + .get_service_statistics_in_interval( + &mixnet_statistics_request.since, + &mixnet_statistics_request.until, + ) + .await + .unwrap() + .into_iter() + .map(|data| StatsMessage { + description: data.service_description, + request_data: StatsData::new(data.request_processed_bytes as u32), + response_data: StatsData::new(data.response_processed_bytes as u32), + interval_seconds: data.interval_seconds as u32, + timestamp: data.timestamp.to_string(), + }) + .collect(); + + Json(MixnetStatisticsResponse { + data: mixnet_statistics, + }) +}