diff --git a/Cargo.lock b/Cargo.lock index a247ea2..3564886 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "android_system_properties" version = "0.1.6" @@ -11,6 +17,66 @@ dependencies = [ "libc", ] +[[package]] +name = "arc-swap" +version = "1.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "380962b3d93b99642e50b9febf72c071e97e8b32d86eae9cf4ef582bb960ae6e" +dependencies = [ + "rustversion", +] + +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure 0.13.2", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "async-compression" +version = "0.4.51" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d5755640120b24961f8231e05892e4a4f7e0d04e8aeca161508c9c397946970" +dependencies = [ + "compression-codecs", + "compression-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -23,6 +89,29 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "aws-lc-rs" +version = "1.18.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "axum" version = "0.8.9" @@ -75,6 +164,49 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-server" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1ab4a3ec9ea8a657c72d99a03a824af695bd0fb5ec639ccbd9cd3543b41a5f9" +dependencies = [ + "arc-swap", + "bytes", + "fs-err", + "http", + "http-body", + "hyper", + "hyper-util", + "pin-project-lite", + "rustls", + "rustls-pemfile", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tower-service", +] + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "2.13.2" @@ -100,6 +232,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86e055fa6e3e40ad6187ba25b6855eeb635d5271f8e0f0e1e9a5826b8d75bea0" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -109,6 +243,23 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + [[package]] name = "chrono" version = "0.4.45" @@ -123,12 +274,105 @@ dependencies = [ "windows-link", ] +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + +[[package]] +name = "compression-codecs" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe" +dependencies = [ + "compression-core", + "flate2", + "memchr", +] + +[[package]] +name = "compression-core" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ebdda89edc534bf8d5dd2857562633b6495557e5846552ade994b99ed13c36a" + [[package]] name = "core-foundation-sys" version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + +[[package]] +name = "crc32fast" +version = "1.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ba106e844045ddc02c20768ac4b0b7cb45f2cc2ecbd833ac9f5dacb183c2055" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + +[[package]] +name = "deranged" +version = "0.5.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e9de72ce2ad1f90dc62fa25f0f430ef85eb4b0d8fa0be4f30373bc40a21d28e" + +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + +[[package]] +name = "equivalent" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826" + [[package]] name = "fallible-iterator" version = "0.3.0" @@ -147,6 +391,23 @@ version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" +[[package]] +name = "flate2" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" +dependencies = [ + "crc32fast", + "miniz_oxide", + "zlib-rs", +] + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -162,12 +423,43 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs-err" +version = "3.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5c95b673b8f6f7235229ae11c5642d81b04c2e64c1e2fb417bc0cf73ca45f29" +dependencies = [ + "autocfg", + "tokio", +] + +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + [[package]] name = "futures-core" version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -186,6 +478,19 @@ dependencies = [ "slab", ] +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "wasi", + "wasm-bindgen", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -193,8 +498,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi", + "rand_core", + "wasm-bindgen", +] + +[[package]] +name = "h2" +version = "0.4.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d29020232d6aa3fb1daca64c1127cf662cf97f254ae16c18c05b8ab635fc118" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", ] [[package]] @@ -268,7 +595,9 @@ checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" dependencies = [ "atomic-waker", "bytes", + "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", @@ -277,6 +606,23 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfa8e654703247911e29c23fbeaa261834bd9bb74efba2f9acddc37bfb127f53" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", ] [[package]] @@ -285,13 +631,22 @@ version = "0.1.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" dependencies = [ + "base64 0.23.1", "bytes", + "futures-channel", + "futures-util", "http", "http-body", + "httparse", "hyper", + "ipnet", + "libc", + "percent-encoding", "pin-project-lite", + "socket2", "tokio", "tower-service", + "tracing", ] [[package]] @@ -318,12 +673,142 @@ dependencies = [ "cc", ] +[[package]] +name = "icu_collections" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe090a42a0d3140fed8c63263ed1fecc02422cb2f4bfe55cd8de2de1b8e105dd" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25d29b57c17c6528d0dc4c2141cfa2d08200465acb537a3909c8b9b8a4a984c0" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "404a3b8c486f205c87fe7f23ab6894d84a4ad76d6e4671ae1981ba7def87c599" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20f0088bfb7441f7a37fa0c9e20cb1e9cba1cac6b9ce8cd2a6dc44294f8e8cbc" + +[[package]] +name = "icu_properties" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ea83d991c56158f7cf898bf86436c1dfe7c07ee82fe88951f2695c9f37a379f" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b98f29af82d02b163fe50161388350e743bb3107cc8e8c2f7314cb505faddfcf" + +[[package]] +name = "icu_provider" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413099315636c588ee8f011190b329b971a09a8fae9d157e01dcae8379468514" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "indexmap" +version = "2.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.106" @@ -358,12 +843,24 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "log" version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "lru-slab" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" + [[package]] name = "matchit" version = "0.8.4" @@ -382,6 +879,22 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.4" @@ -390,7 +903,17 @@ checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", ] [[package]] @@ -399,7 +922,32 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", ] [[package]] @@ -411,12 +959,31 @@ dependencies = [ "autocfg", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "pem" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d354a98a3d1251555de99e8fdd8afda05573c31b82f59063a7b0a29b5527f120" +dependencies = [ + "base64 0.23.1", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -435,6 +1002,21 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + +[[package]] +name = "powerfmt" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a6394b9e965e73d0a289ee54f589087e2c676aedf60885baf52c76b771e4958" + [[package]] name = "proc-macro2" version = "1.0.107" @@ -444,6 +1026,62 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quinn" +version = "0.11.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af66907df18639dcf4db56ca65490cabc4b27a97dbadd96f2926cca73298f016" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -459,6 +1097,98 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core", +] + +[[package]] +name = "rcgen" +version = "0.14.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http 0.6.11", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rsqlite-vfs" version = "0.2.0" @@ -488,16 +1218,85 @@ name = "rust_test" version = "0.1.0" dependencies = [ "axum", + "axum-server", "chrono", + "rcgen", + "reqwest", "rusqlite", + "rustls", + "rustls-pemfile", + "rustls-pki-types", "serde", "serde_json", "tokio", + "tower-http 0.7.1", "tracing", "tracing-subscriber", "uuid", ] +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + +[[package]] +name = "rustls" +version = "0.23.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +dependencies = [ + "aws-lc-rs", + "log", + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pemfile" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" +dependencies = [ + "aws-lc-rs", + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" @@ -591,6 +1390,12 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "slab" version = "0.4.12" @@ -610,7 +1415,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -623,6 +1428,18 @@ dependencies = [ "rsqlite-vfs", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -650,6 +1467,31 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "synstructure" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] [[package]] name = "thiserror" @@ -680,18 +1522,65 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + +[[package]] +name = "time-macros" +version = "0.2.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85" +dependencies = [ + "num-conv", + "time-core", +] + +[[package]] +name = "tinystr" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abc840c238154d58f9e86b2792261edfde565a17696e1cda1a449de27979801a" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + [[package]] name = "tokio" version = "1.54.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" dependencies = [ + "bytes", "libc", "mio", "pin-project-lite", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -705,6 +1594,30 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "tower" version = "0.5.3" @@ -721,6 +1634,49 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "async-compression", + "bitflags", + "bytes", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "pin-project-lite", + "tokio", + "tokio-util", + "tower", + "tower-layer", + "tower-service", + "url", +] + +[[package]] +name = "tower-http" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c" +dependencies = [ + "async-compression", + "bitflags", + "bytes", + "futures-core", + "http", + "http-body", + "percent-encoding", + "pin-project-lite", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -791,19 +1747,49 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "unicode-ident" version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "uuid" version = "1.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4a0e2d11620c50695fe12bbbfde96b3477fe606e7284c22d505c12ce51214514" dependencies = [ - "getrandom", + "getrandom 0.4.3", "js-sys", "serde_core", "wasm-bindgen", @@ -821,6 +1807,15 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" +[[package]] +name = "want" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -840,6 +1835,17 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.79" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cbab34de2d982e9b48e18d216d04c4a6f641066ff19ffb699980f591ee3610e" +dependencies = [ + "js-sys", + "tokio", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.129" @@ -872,6 +1878,35 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-sys" +version = "0.3.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88261b9deccee56594c11a3460c462c41f58d148598fe70ad77070126a68aba4" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "webpki-roots" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -931,6 +1966,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -940,6 +1984,193 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] + +[[package]] +name = "yoke" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "417ef3fcf6c300b7dbd25c85531a746931c02ad536f3cf4786ed1e88223340b3" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", + "synstructure 0.14.0", +] + +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", + "synstructure 0.14.0", +] + +[[package]] +name = "zeroize" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13084392c5e4bc371903e2935a5eaeed24905a7511356b883835e18a78f6879" + +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f58a2799cfe9a1670c7ea8edbd52ef8edb93cb37a7f83bf31ba9e35d9dae6a40" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3743ddbf47738d929cd1ae2e199d40d0162979dfe31aef8ae646d2bf00bf97a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "zlib-rs" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" + [[package]] name = "zmij" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index e7223c8..0816f19 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,9 +5,15 @@ edition = "2024" [dependencies] axum = { version = "0.8.9", features = ["json"] } +axum-server = { version = "0.7", features = ["tls-rustls"] } chrono = { version = "0.4.45", features = ["serde"] } +rcgen = { version = "0.14.10", features = ["pem", "x509-parser"] } reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "gzip"] } rusqlite = { version = "0.41.0", features = ["bundled"] } +rustls = { version = "0.23.46", features = ["ring"] } +rustls-pki-types = "1.0" +rustls-pemfile = "2.1" + serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.154" tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] } diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..ff98980 --- /dev/null +++ b/src/db.rs @@ -0,0 +1,83 @@ +use rusqlite::{params, Connection, OptionalExtension}; +use tracing::{info, warn}; +use uuid::Uuid; + +pub fn meta_get(db: &Connection, key: &str) -> rusqlite::Result> { + db.query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| { + r.get::<_, String>(0) + }) + .optional() +} + +pub fn meta_set(db: &Connection, key: &str, value: &str) -> rusqlite::Result<()> { + db.execute( + "INSERT INTO meta (key, value) VALUES (?1, ?2) + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![key, value], + )?; + Ok(()) +} + +pub fn init_db(db: &mut Connection) -> rusqlite::Result<()> { + db.execute_batch( + "CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + );", + )?; + + let exists: bool = db.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'records')", + [], + |r| r.get(0), + )?; + + if exists { + let has_seq: bool = db.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info('records') WHERE name = 'seq')", + [], + |r| r.get(0), + )?; + + if !has_seq { + info!("Migrating records table to journal layout"); + let tx = db.transaction()?; + tx.execute_batch( + "ALTER TABLE records RENAME TO records_old; + CREATE TABLE records ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + date TEXT NOT NULL, + data TEXT NOT NULL + ); + INSERT INTO records (uuid, date, data) + SELECT uuid, date, data FROM records_old ORDER BY rowid; + DROP TABLE records_old;", + )?; + tx.commit()?; + } + } + + db.execute_batch( + "CREATE TABLE IF NOT EXISTS records ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + date TEXT NOT NULL, + data TEXT NOT NULL + );", + )?; + + Ok(()) +} + +pub fn load_or_create_instance_id(db: &Connection) -> rusqlite::Result { + if let Some(existing) = meta_get(db, "instance_id")? { + if let Ok(id) = Uuid::parse_str(&existing) { + return Ok(id); + } + warn!("Stored instance_id is invalid; generating a new one"); + } + let id = Uuid::new_v4(); + meta_set(db, "instance_id", &id.to_string())?; + Ok(id) +} \ No newline at end of file diff --git a/src/handlers.rs b/src/handlers.rs new file mode 100644 index 0000000..24aa424 --- /dev/null +++ b/src/handlers.rs @@ -0,0 +1,287 @@ +use axum::{ + extract::{rejection::JsonRejection, State}, + http::StatusCode, + Json, +}; +use chrono::Utc; +use crate::tls::{create_join_token, issue_leaf_cert, JoinClusterRequest, JoinClusterResponse}; + +use rusqlite::params; +use serde_json::Value; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use crate::{ + schema::{load_config, validate_data, ValidationMode}, + state::{api_error, db_error, internal_error, ApiError, ApiResponse, AppState, CreateRequest, Role}, +}; + +/// Public endpoint: fetch the cluster's public CA certificate. +pub async fn get_cluster_ca( + State(state): State, +) -> Result, ApiError> { + let ca_pem = state + .ca_cert_pem + .as_deref() + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "Cluster CA not available"))?; + + Ok(Json(serde_json::json!({ + "instance_id": state.instance_id, + "ca_pem": ca_pem + }))) +} + +/// Admin endpoint: mint a short-lived join token (valid for 10 minutes by default). +pub async fn create_token_handler( + State(state): State, + payload: Option>, +) -> Result, ApiError> { + if state.role == Role::Replica { + return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can generate join tokens")); + } + + let ttl = payload + .and_then(|p| p.get("ttl_secs").and_then(|v| v.as_u64())) + .unwrap_or(600); + + let token = create_join_token(ttl); + + let mut tokens = state + .join_tokens + .write() + .map_err(|_| internal_error("tokens lock poisoned"))?; + tokens.insert(token.token.clone(), token.clone()); + + info!(token = %token.token, expires_at = %token.expires_at, "Issued join token"); + Ok(Json(serde_json::json!(token))) +} + +/// Dynamic join endpoint: exchanges a valid join token for an mTLS certificate. +pub async fn join_cluster_handler( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + if state.role == Role::Replica { + return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can issue certificates")); + } + + let ca_cert_pem = state + .ca_cert_pem + .as_ref() + .ok_or_else(|| internal_error("CA cert unavailable"))?; + let ca_key_pem = state + .ca_key_pem + .as_ref() + .ok_or_else(|| internal_error("CA key unavailable"))?; + + // 1. Validate & burn the token + { + let mut tokens = state + .join_tokens + .write() + .map_err(|_| internal_error("tokens lock poisoned"))?; + + let entry = tokens + .get_mut(&req.token) + .ok_or_else(|| api_error(StatusCode::UNAUTHORIZED, "Invalid join token"))?; + + if entry.used { + return Err(api_error(StatusCode::UNAUTHORIZED, "Join token already used")); + } + if entry.expires_at < Utc::now() { + return Err(api_error(StatusCode::UNAUTHORIZED, "Join token expired")); + } + + entry.used = true; + } + + // 2. Issue leaf certificate signed by Cluster CA + let mut sans = req.san_list; + if !sans.iter().any(|s| s == "localhost") { + sans.push("localhost".into()); + } + if !sans.iter().any(|s| s == "127.0.0.1") { + sans.push("127.0.0.1".into()); + } + + let (cert_pem, key_pem) = issue_leaf_cert( + ca_cert_pem, + ca_key_pem, + &req.node_id.to_string(), + sans, + ) + .map_err(|e| internal_error(&format!("cert issue failed: {e}")))?; + + info!(node_id = %req.node_id, "Issued client/server mTLS cert to replica"); + + Ok(Json(JoinClusterResponse { + ca_pem: ca_cert_pem.to_string(), + cert_pem, + key_pem, + })) +} + + +pub async fn get_info() -> Json { + info!(endpoint = "/api/now", "Request received"); + Json(ApiResponse { + seq: None, + uuid: Uuid::new_v4(), + date: Utc::now().to_rfc3339(), + data: serde_json::json!({ + "version": 1, + "message": "Hello from Rust!" + }), + }) +} + +pub async fn create_record( + State(state): State, + payload: Result, JsonRejection>, +) -> Result<(StatusCode, Json), ApiError> { + info!(endpoint = "/api/records", "Create request received"); + + if state.role == Role::Replica { + return Err(api_error( + StatusCode::FORBIDDEN, + format!( + "This instance is a read-only replica; write to {}", + state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary") + ), + )); + } + + let Json(request) = payload.map_err(|e| { + warn!(error = %e, "Invalid JSON request"); + api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body") + })?; + + let schema = state + .schema + .read() + .map_err(|_| internal_error("schema lock poisoned"))? + .clone(); + + if let Err(message) = validate_data(&request.data, &schema) { + warn!(reason = %message, "Schema validation failed"); + return Err(api_error(StatusCode::BAD_REQUEST, message)); + } + + let mut record = ApiResponse { + seq: None, + uuid: Uuid::new_v4(), + date: Utc::now().to_rfc3339(), + data: request.data, + }; + + let data_json = serde_json::to_string(&record.data) + .map_err(|_| internal_error("serialize data"))?; + + { + let db = state + .db + .lock() + .map_err(|_| internal_error("database lock poisoned"))?; + + db.execute( + "INSERT INTO records (uuid, date, data) VALUES (?1, ?2, ?3)", + params![record.uuid.to_string(), record.date, data_json], + ) + .map_err(|e| { + error!(error = %e, "Failed to insert record"); + api_error(StatusCode::INTERNAL_SERVER_ERROR, "Failed to store record") + })?; + + record.seq = Some(db.last_insert_rowid()); + } + + info!(uuid = %record.uuid, seq = ?record.seq, "Record stored"); + Ok((StatusCode::CREATED, Json(record))) +} + +pub async fn list_records( + State(state): State, +) -> Result>, ApiError> { + info!(endpoint = "/api/records", "List request received"); + + let db = state + .db + .lock() + .map_err(|_| internal_error("database lock poisoned"))?; + + let mut statement = db + .prepare("SELECT seq, uuid, date, data FROM records ORDER BY seq DESC LIMIT 100") + .map_err(|e| db_error("prepare list", e))?; + + let rows = statement + .query_map([], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + )) + }) + .map_err(|e| db_error("query list", e))?; + + let mut records = Vec::new(); + for row in rows { + let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read row", e))?; + let uuid = Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?; + let data: Value = + serde_json::from_str(&data_json).map_err(|_| internal_error("invalid stored JSON"))?; + + records.push(ApiResponse { + seq: Some(seq), + uuid, + date, + data, + }); + } + + info!(count = records.len(), "Records returned"); + Ok(Json(records)) +} + +pub async fn get_config(State(state): State) -> Result, ApiError> { + let schema = state + .schema + .read() + .map_err(|_| internal_error("schema lock poisoned"))? + .clone(); + + Ok(Json(schema)) +} + +pub async fn reload_config(State(state): State) -> Result, ApiError> { + info!(path = %state.config_path, "Configuration reload requested"); + + let new_schema = match load_config(&state.config_path) { + Ok(config) => config, + Err(message) => { + error!(reason = %message, "Configuration reload failed"); + return Err(api_error(StatusCode::UNPROCESSABLE_ENTITY, message)); + } + }; + + let mode = match new_schema.mode { + ValidationMode::Strict => "strict", + ValidationMode::Loose => "loose", + }; + let field_count = new_schema.fields.len(); + + { + let mut active = state + .schema + .write() + .map_err(|_| internal_error("schema lock poisoned"))?; + *active = new_schema; + } + + info!(mode, fields = field_count, "Configuration reloaded successfully"); + Ok(Json(serde_json::json!({ + "status": "ok", + "mode": mode, + "fields": field_count + }))) +} diff --git a/src/main.rs b/src/main.rs index 22d97f9..26fc4d2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,913 +1,43 @@ +mod db; +mod handlers; +mod replication; +mod schema; +mod state; +mod tls; + use axum::{ - extract::{rejection::JsonRejection, Query, State}, http::{ header::{AUTHORIZATION, CONTENT_TYPE}, - HeaderMap, HeaderValue, Method, StatusCode, + HeaderValue, Method, }, - routing::get, - Json, Router, + routing::{get, post}, + Router, }; -use chrono::Utc; -use rusqlite::{params, Connection, OptionalExtension}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; +use axum_server::tls_rustls::RustlsConfig; +use rusqlite::Connection; use std::{ - fs, + collections::HashMap, + net::SocketAddr, sync::{Arc, Mutex, RwLock}, time::Duration, }; -use tracing::{error, info, warn}; +use tracing::{info, warn}; use uuid::Uuid; -// --------------------------------------------------------------------------- -// State -// --------------------------------------------------------------------------- - -#[derive(Clone)] -struct AppState { - db: Arc>, - schema: Arc>, - config_path: Arc, - - /// Permanent per-instance identity (stored in the `meta` table). - instance_id: Uuid, - role: Role, - primary_url: Option>, - /// Shared secret for /api/replication/*. None = replication endpoint disabled. - repl_token: Option>, - sync: Arc>, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum Role { - Primary, - Replica, -} - -impl Role { - fn as_str(self) -> &'static str { - match self { - Role::Primary => "primary", - Role::Replica => "replica", - } - } -} - -#[derive(Debug, Clone, Serialize)] -struct SyncStatus { - /// "primary" | "idle" | "syncing" | "retrying" | "halted" - state: String, - /// Last global tick (seq) applied locally. Replica's cursor. - last_applied_seq: i64, - source_instance_id: Option, - /// Primary's head seq as of the last successful pull. - primary_head_seq: Option, - last_success: Option, - last_error: Option, -} - -// --------------------------------------------------------------------------- -// Schema validation (unchanged) -// --------------------------------------------------------------------------- - -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] -#[serde(rename_all = "lowercase")] -enum ValidationMode { - Strict, - Loose, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -struct FieldSpec { - #[serde(rename = "type")] - data_type: DataType, - - #[serde(default, skip_serializing_if = "Option::is_none")] - value: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "lowercase")] -enum DataType { - String, - Integer, - Number, - Boolean, - Object, - Array, - Null, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -struct SchemaConfig { - mode: ValidationMode, - fields: std::collections::HashMap, -} - -#[derive(Debug, Serialize, Deserialize)] -struct ApiResponse { - /// Global tick. Absent on endpoints that aren't records. - #[serde(default, skip_serializing_if = "Option::is_none")] - seq: Option, - uuid: Uuid, - date: String, - data: Value, -} - -#[derive(Debug, Deserialize)] -struct CreateRequest { - data: Value, -} - -type ApiError = (StatusCode, String); - -fn api_error(status: StatusCode, message: impl Into) -> ApiError { - (status, message.into()) -} - -fn internal_error(context: &str) -> ApiError { - error!(context, "Internal server error"); - api_error( - StatusCode::INTERNAL_SERVER_ERROR, - "Internal server error", - ) -} - -fn db_error(context: &str, e: rusqlite::Error) -> ApiError { - error!(error = %e, context, "Database error"); - api_error(StatusCode::INTERNAL_SERVER_ERROR, "Database error") -} - -fn matches_type(value: &Value, expected: &DataType) -> bool { - match expected { - DataType::String => value.is_string(), - DataType::Integer => { - value.as_i64().is_some() || value.as_u64().is_some() - } - DataType::Number => value.is_number(), - DataType::Boolean => value.is_boolean(), - DataType::Object => value.is_object(), - DataType::Array => value.is_array(), - DataType::Null => value.is_null(), - } -} - -fn type_name(data_type: &DataType) -> &'static str { - match data_type { - DataType::String => "string", - DataType::Integer => "integer", - DataType::Number => "number", - DataType::Boolean => "boolean", - DataType::Object => "object", - DataType::Array => "array", - DataType::Null => "null", - } -} - -fn validate_data(data: &Value, schema: &SchemaConfig) -> Result<(), String> { - let object = data - .as_object() - .ok_or("data must be a JSON object")?; - - for (key, spec) in &schema.fields { - let value = object - .get(key) - .ok_or_else(|| format!("Missing required field: {key}"))?; - - if !matches_type(value, &spec.data_type) { - return Err(format!( - "Field '{key}' must be of type {}", - type_name(&spec.data_type) - )); - } - - if let Some(expected) = &spec.value { - if value != expected { - return Err(format!( - "Field '{key}' must equal {}", - expected - )); - } - } - } - - if matches!(schema.mode, ValidationMode::Strict) { - for key in object.keys() { - if !schema.fields.contains_key(key) { - return Err(format!("Unknown field: {key}")); - } - } - } - - Ok(()) -} - -fn load_config(path: &str) -> Result { - let contents = fs::read_to_string(path) - .map_err(|e| format!("Cannot read {path}: {e}"))?; - - let config: SchemaConfig = serde_json::from_str(&contents) - .map_err(|e| format!("Invalid configuration JSON: {e}"))?; - - if config.fields.is_empty() { - return Err("Configuration must define at least one field".into()); - } - - Ok(config) -} - -// --------------------------------------------------------------------------- -// Database init / meta -// --------------------------------------------------------------------------- - -fn meta_get(db: &Connection, key: &str) -> rusqlite::Result> { - db.query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| { - r.get::<_, String>(0) - }) - .optional() -} - -fn meta_set(db: &Connection, key: &str, value: &str) -> rusqlite::Result<()> { - db.execute( - "INSERT INTO meta (key, value) VALUES (?1, ?2) - ON CONFLICT(key) DO UPDATE SET value = excluded.value", - params![key, value], - )?; - Ok(()) -} - -/// Creates tables and migrates a pre-replication `records` table -/// (uuid PK, no seq) to the journal layout (seq AUTOINCREMENT + uuid UNIQUE). -/// -/// `records` is insert-only, so it *is* the append-only journal; `seq` is the -/// global tick. AUTOINCREMENT guarantees seq never gets reused, even after -/// a rollback. -fn init_db(db: &mut Connection) -> rusqlite::Result<()> { - db.execute_batch( - "CREATE TABLE IF NOT EXISTS meta ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL - );", - )?; - - let exists: bool = db.query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_master - WHERE type = 'table' AND name = 'records')", - [], - |r| r.get(0), - )?; - - if exists { - let has_seq: bool = db.query_row( - "SELECT EXISTS(SELECT 1 FROM pragma_table_info('records') - WHERE name = 'seq')", - [], - |r| r.get(0), - )?; - - if !has_seq { - info!("Migrating records table to journal layout"); - let tx = db.transaction()?; - tx.execute_batch( - "ALTER TABLE records RENAME TO records_old; - CREATE TABLE records ( - seq INTEGER PRIMARY KEY AUTOINCREMENT, - uuid TEXT NOT NULL UNIQUE, - date TEXT NOT NULL, - data TEXT NOT NULL - ); - INSERT INTO records (uuid, date, data) - SELECT uuid, date, data FROM records_old ORDER BY rowid; - DROP TABLE records_old;", - )?; - tx.commit()?; - } - } - - db.execute_batch( - "CREATE TABLE IF NOT EXISTS records ( - seq INTEGER PRIMARY KEY AUTOINCREMENT, - uuid TEXT NOT NULL UNIQUE, - date TEXT NOT NULL, - data TEXT NOT NULL - );", - )?; - - Ok(()) -} - -fn load_or_create_instance_id(db: &Connection) -> rusqlite::Result { - if let Some(existing) = meta_get(db, "instance_id")? { - if let Ok(id) = Uuid::parse_str(&existing) { - return Ok(id); - } - warn!("Stored instance_id is invalid; generating a new one"); - } - let id = Uuid::new_v4(); - meta_set(db, "instance_id", &id.to_string())?; - Ok(id) -} - -// --------------------------------------------------------------------------- -// Regular handlers -// --------------------------------------------------------------------------- - -async fn get_info() -> Json { - info!(endpoint = "/api/now", "Request received"); - - Json(ApiResponse { - seq: None, - uuid: Uuid::new_v4(), - date: Utc::now().to_rfc3339(), - data: serde_json::json!({ - "version": 1, - "message": "Hello from Rust!" - }), - }) -} - -async fn create_record( - State(state): State, - payload: Result, JsonRejection>, -) -> Result<(StatusCode, Json), ApiError> { - info!(endpoint = "/api/records", "Create request received"); - - if state.role == Role::Replica { - return Err(api_error( - StatusCode::FORBIDDEN, - format!( - "This instance is a read-only replica; write to {}", - state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary") - ), - )); - } - - let Json(request) = payload.map_err(|e| { - warn!(error = %e, "Invalid JSON request"); - api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body") - })?; - - // Take a snapshot; reloads won't affect this request mid-validation. - let schema = state - .schema - .read() - .map_err(|_| internal_error("schema lock poisoned"))? - .clone(); - - if let Err(message) = validate_data(&request.data, &schema) { - warn!(reason = %message, "Schema validation failed"); - return Err(api_error(StatusCode::BAD_REQUEST, message)); - } - - let mut record = ApiResponse { - seq: None, - uuid: Uuid::new_v4(), - date: Utc::now().to_rfc3339(), - data: request.data, - }; - - let data_json = serde_json::to_string(&record.data) - .map_err(|_| internal_error("serialize data"))?; - - { - let db = state - .db - .lock() - .map_err(|_| internal_error("database lock poisoned"))?; - - db.execute( - "INSERT INTO records (uuid, date, data) - VALUES (?1, ?2, ?3)", - params![ - record.uuid.to_string(), - record.date, - data_json, - ], - ) - .map_err(|e| { - error!(error = %e, "Failed to insert record"); - api_error( - StatusCode::INTERNAL_SERVER_ERROR, - "Failed to store record", - ) - })?; - - record.seq = Some(db.last_insert_rowid()); - } - - info!(uuid = %record.uuid, seq = ?record.seq, "Record stored"); - - Ok((StatusCode::CREATED, Json(record))) -} - -async fn list_records( - State(state): State, -) -> Result>, ApiError> { - info!(endpoint = "/api/records", "List request received"); - - let db = state - .db - .lock() - .map_err(|_| internal_error("database lock poisoned"))?; - - let mut statement = db - .prepare( - "SELECT seq, uuid, date, data - FROM records - ORDER BY seq DESC - LIMIT 100", - ) - .map_err(|e| db_error("prepare list", e))?; - - let rows = statement - .query_map([], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - )) - }) - .map_err(|e| db_error("query list", e))?; - - let mut records = Vec::new(); - - for row in rows { - let (seq, uuid, date, data_json) = - row.map_err(|e| db_error("read row", e))?; - - let uuid = Uuid::parse_str(&uuid) - .map_err(|_| internal_error("invalid stored UUID"))?; - - let data: Value = serde_json::from_str(&data_json) - .map_err(|_| internal_error("invalid stored JSON"))?; - - records.push(ApiResponse { - seq: Some(seq), - uuid, - date, - data, - }); - } - - info!(count = records.len(), "Records returned"); - - Ok(Json(records)) -} - -async fn get_config( - State(state): State, -) -> Result, ApiError> { - let schema = state - .schema - .read() - .map_err(|_| internal_error("schema lock poisoned"))? - .clone(); - - Ok(Json(schema)) -} - -async fn reload_config( - State(state): State, -) -> Result, ApiError> { - info!(path = %state.config_path, "Configuration reload requested"); - - // Parse and validate the new file before touching active configuration. - let new_schema = match load_config(&state.config_path) { - Ok(config) => config, - Err(message) => { - error!(reason = %message, "Configuration reload failed"); - return Err(api_error( - StatusCode::UNPROCESSABLE_ENTITY, - message, - )); - } - }; - - let mode = match new_schema.mode { - ValidationMode::Strict => "strict", - ValidationMode::Loose => "loose", - }; - - let field_count = new_schema.fields.len(); - - { - let mut active = state - .schema - .write() - .map_err(|_| internal_error("schema lock poisoned"))?; - - *active = new_schema; - } - - info!( - mode, - fields = field_count, - "Configuration reloaded successfully" - ); - - Ok(Json(serde_json::json!({ - "status": "ok", - "mode": mode, - "fields": field_count - }))) -} - -// --------------------------------------------------------------------------- -// Replication: serving side (primary, or a replica acting as a relay) -// --------------------------------------------------------------------------- - -#[derive(Debug, Serialize, Deserialize)] -struct Change { - seq: i64, - uuid: Uuid, - date: String, - data: Value, -} - -#[derive(Debug, Serialize, Deserialize)] -struct ChangesResponse { - instance_id: Uuid, - head_seq: i64, - has_more: bool, - changes: Vec, -} - -#[derive(Debug, Deserialize)] -struct ChangesQuery { - after: Option, - limit: Option, -} - -fn ct_eq(a: &[u8], b: &[u8]) -> bool { - a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 -} - -fn check_token( - headers: &HeaderMap, - expected: &Option>, -) -> Result<(), ApiError> { - let Some(expected) = expected else { - return Err(api_error( - StatusCode::FORBIDDEN, - "Replication is disabled (no token configured)", - )); - }; - - let provided = headers - .get(AUTHORIZATION) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.strip_prefix("Bearer ")) - .unwrap_or(""); - - if !ct_eq(provided.as_bytes(), expected.as_bytes()) { - return Err(api_error(StatusCode::UNAUTHORIZED, "Invalid token")); - } - Ok(()) -} - -/// Cursor-based pull. Stateless on the server: the replica says "give me -/// everything after seq N", so dropped connections, duplicate requests and -/// restarts on either side are all harmless. -async fn replication_changes( - State(state): State, - headers: HeaderMap, - Query(query): Query, -) -> Result, ApiError> { - check_token(&headers, &state.repl_token)?; - - let after = query.after.unwrap_or(0).max(0); - let limit = query.limit.unwrap_or(200).clamp(1, 1000); - - let db = state - .db - .lock() - .map_err(|_| internal_error("database lock poisoned"))?; - - let head_seq: i64 = db - .query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { - r.get(0) - }) - .map_err(|e| db_error("head seq", e))?; - - let mut statement = db - .prepare( - "SELECT seq, uuid, date, data - FROM records - WHERE seq > ?1 - ORDER BY seq ASC - LIMIT ?2", - ) - .map_err(|e| db_error("prepare changes", e))?; - - let rows = statement - .query_map(params![after, limit], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - )) - }) - .map_err(|e| db_error("query changes", e))?; - - let mut changes = Vec::new(); - for row in rows { - let (seq, uuid, date, data_json) = - row.map_err(|e| db_error("read change", e))?; - changes.push(Change { - seq, - uuid: Uuid::parse_str(&uuid) - .map_err(|_| internal_error("invalid stored UUID"))?, - date, - data: serde_json::from_str(&data_json) - .map_err(|_| internal_error("invalid stored JSON"))?, - }); - } - - let last = changes.last().map(|c| c.seq).unwrap_or(after); - - Ok(Json(ChangesResponse { - instance_id: state.instance_id, - head_seq, - has_more: last < head_seq, - changes, - })) -} - -async fn replication_status( - State(state): State, -) -> Result, ApiError> { - let status = state - .sync - .read() - .map_err(|_| internal_error("sync status lock poisoned"))? - .clone(); - - let local_head: i64 = { - let db = state - .db - .lock() - .map_err(|_| internal_error("database lock poisoned"))?; - db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { - r.get(0) - }) - .map_err(|e| db_error("head seq", e))? - }; - - let lag = status - .primary_head_seq - .map(|head| (head - status.last_applied_seq).max(0)); - - Ok(Json(serde_json::json!({ - "instance_id": state.instance_id, - "role": state.role.as_str(), - "primary_url": state.primary_url.as_deref(), - "local_head_seq": local_head, - "lag_records": lag, - "sync": status, - }))) -} - -// --------------------------------------------------------------------------- -// Replication: replica side -// --------------------------------------------------------------------------- - -#[derive(Debug, Clone)] -struct SyncConfig { - poll_interval: Duration, - max_backoff: Duration, - batch: i64, -} - -enum SyncError { - /// Network trouble, 5xx, truncated body... retry with backoff. - Transient(String), - /// Identity/history mismatch. Retrying won't help; needs an operator. - Fatal(String), -} - -fn jitter(d: Duration) -> Duration { - // Uniform in [d/2, d]; avoids a fleet of replicas retrying in lockstep. - let ms = d.as_millis() as u64; - let extra = (Uuid::new_v4().as_u128() as u64) % (ms / 2 + 1); - Duration::from_millis(ms / 2 + extra) -} - -fn update_status(state: &AppState, f: impl FnOnce(&mut SyncStatus)) { - if let Ok(mut s) = state.sync.write() { - f(&mut s); - } -} - -/// Applies a batch atomically: rows + cursor + source identity commit together, -/// so a crash can never leave the cursor ahead of (or behind) the data. -/// INSERT OR IGNORE makes re-delivery idempotent. -fn apply_batch( - db: &Arc>, - source: Uuid, - changes: &[Change], -) -> Result { - let mut conn = db.lock().map_err(|_| "database lock poisoned".to_string())?; - let tx = conn.transaction().map_err(|e| e.to_string())?; - - let mut max_seq = 0; - { - let mut insert = tx - .prepare_cached( - "INSERT OR IGNORE INTO records (seq, uuid, date, data) - VALUES (?1, ?2, ?3, ?4)", - ) - .map_err(|e| e.to_string())?; - - for c in changes { - let data = serde_json::to_string(&c.data).map_err(|e| e.to_string())?; - insert - .execute(params![c.seq, c.uuid.to_string(), c.date, data]) - .map_err(|e| e.to_string())?; - max_seq = max_seq.max(c.seq); - } - } - - meta_set(&tx, "last_applied_seq", &max_seq.to_string()) - .map_err(|e| e.to_string())?; - meta_set(&tx, "source_instance_id", &source.to_string()) - .map_err(|e| e.to_string())?; - - tx.commit().map_err(|e| e.to_string())?; - Ok(max_seq) -} - -async fn sync_once( - client: &reqwest::Client, - state: &AppState, - primary_url: &str, - token: &str, - batch: i64, -) -> Result { - let (last, known_source) = { - let s = state - .sync - .read() - .map_err(|_| SyncError::Fatal("sync status lock poisoned".into()))?; - (s.last_applied_seq, s.source_instance_id) - }; - - let url = format!( - "{}/api/replication/changes", - primary_url.trim_end_matches('/') - ); - - let resp = client - .get(&url) - .bearer_auth(token) - .query(&[("after", last), ("limit", batch)]) - .send() - .await - .map_err(|e| SyncError::Transient(format!("request failed: {e}")))?; - - if !resp.status().is_success() { - return Err(SyncError::Transient(format!( - "primary returned {}", - resp.status() - ))); - } - - // A truncated or garbled body on a flaky link fails to decode here and is - // retried; a partial batch is never applied. - let body: ChangesResponse = resp - .json() - .await - .map_err(|e| SyncError::Transient(format!("bad response body: {e}")))?; - - // --- identity / history checks --- - if body.instance_id == state.instance_id { - return Err(SyncError::Fatal( - "source has the same instance_id as this replica (cloned database?)".into(), - )); - } - if let Some(known) = known_source { - if known != body.instance_id { - return Err(SyncError::Fatal(format!( - "source instance changed: expected {known}, got {}", - body.instance_id - ))); - } - } - if body.head_seq < last { - return Err(SyncError::Fatal(format!( - "source head_seq {} is behind our cursor {last} (source restored from older backup?)", - body.head_seq - ))); - } - - if body.changes.is_empty() { - update_status(state, |s| { - s.state = "idle".into(); - s.primary_head_seq = Some(body.head_seq); - s.last_success = Some(Utc::now().to_rfc3339()); - s.last_error = None; - s.source_instance_id = Some(body.instance_id); - }); - return Ok(false); - } - - // Sequence must be strictly increasing and past our cursor. - let mut prev = last; - for c in &body.changes { - if c.seq <= prev { - return Err(SyncError::Transient(format!( - "non-monotonic seq {} after {prev}", - c.seq - ))); - } - prev = c.seq; - } - - let count = body.changes.len(); - let db = state.db.clone(); - let source = body.instance_id; - let changes = body.changes; - - let max_seq = tokio::task::spawn_blocking(move || apply_batch(&db, source, &changes)) - .await - .map_err(|e| SyncError::Transient(format!("apply task failed: {e}")))? - .map_err(|e| SyncError::Transient(format!("apply failed: {e}")))?; - - update_status(state, |s| { - s.state = if body.has_more { "syncing" } else { "idle" }.into(); - s.last_applied_seq = max_seq; - s.source_instance_id = Some(source); - s.primary_head_seq = Some(body.head_seq); - s.last_success = Some(Utc::now().to_rfc3339()); - s.last_error = None; - }); - - info!( - applied = count, - last_applied_seq = max_seq, - head = body.head_seq, - "Replication batch applied" - ); - - Ok(body.has_more) -} - -async fn replica_loop( - state: AppState, - primary_url: String, - token: String, - cfg: SyncConfig, -) { - let client = match reqwest::Client::builder() - .connect_timeout(Duration::from_secs(10)) - .timeout(Duration::from_secs(60)) - .build() - { - Ok(c) => c, - Err(e) => { - error!(error = %e, "Cannot build HTTP client; replication halted"); - return; - } - }; - - let mut backoff = Duration::from_secs(1); - - info!(primary = %primary_url, "Replica sync loop started"); - - loop { - match sync_once(&client, &state, &primary_url, &token, cfg.batch).await { - Ok(more) => { - backoff = Duration::from_secs(1); - if !more { - tokio::time::sleep(jitter(cfg.poll_interval)).await; - } - } - Err(SyncError::Transient(message)) => { - let wait = jitter(backoff); - warn!(reason = %message, retry_in = ?wait, "Replication pull failed"); - update_status(&state, |s| { - s.state = "retrying".into(); - s.last_error = Some(message); - }); - tokio::time::sleep(wait).await; - backoff = (backoff * 2).min(cfg.max_backoff); - } - Err(SyncError::Fatal(message)) => { - error!(reason = %message, "Replication halted; operator action required"); - update_status(&state, |s| { - s.state = "halted".into(); - s.last_error = Some(message); - }); - return; - } - } - } -} - -// --------------------------------------------------------------------------- -// main -// --------------------------------------------------------------------------- +use crate::{ + db::{init_db, load_or_create_instance_id, meta_get, meta_set}, + handlers::{ + create_record, create_token_handler, get_cluster_ca, get_config, get_info, + join_cluster_handler, list_records, reload_config, + }, + replication::{replica_loop, replication_changes, replication_status, SyncConfig}, + schema::load_config, + state::{AppState, Role, SyncStatus}, + tls::{ + bootstrap_replica_join, create_mtls_server_config, generate_ca, issue_leaf_cert, + JoinToken, + }, +}; fn env_parse(name: &str, default: T) -> T { std::env::var(name) @@ -918,24 +48,18 @@ fn env_parse(name: &str, default: T) -> T { #[tokio::main] async fn main() { + let _ = rustls::crypto::ring::default_provider().install_default(); + tracing_subscriber::fmt() .with_writer(std::io::stdout) .with_target(false) .init(); - let config_path = - std::env::var("APP_CONFIG").unwrap_or_else(|_| "config.json".into()); + let config_path = std::env::var("APP_CONFIG").unwrap_or_else(|_| "config.json".into()); + let schema = load_config(&config_path).unwrap_or_else(|e| panic!("Failed to load configuration: {e}")); - let schema = load_config(&config_path) - .unwrap_or_else(|e| panic!("Failed to load configuration: {e}")); + info!(path = %config_path, fields = schema.fields.len(), "Configuration loaded"); - info!( - path = %config_path, - fields = schema.fields.len(), - "Configuration loaded" - ); - - // Replication settings let role = match std::env::var("APP_ROLE").as_deref() { Ok("replica") => Role::Replica, Ok("primary") | Err(_) => Role::Primary, @@ -957,13 +81,78 @@ async fn main() { batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000), }; - let mut db = Connection::open("messages.db") - .expect("Failed to open SQLite database"); - + let mut db = Connection::open("messages.db").expect("Failed to open SQLite database"); init_db(&mut db).expect("Failed to initialize database"); - let instance_id = - load_or_create_instance_id(&db).expect("Failed to load instance id"); + let instance_id = load_or_create_instance_id(&db).expect("Failed to load instance id"); + + // ----------------------------------------------------------------------- + // mTLS Credential Initialization & Dynamic Join + // ----------------------------------------------------------------------- + let mut ca_cert_pem = meta_get(&db, "tls_ca_pem").ok().flatten(); + let mut ca_key_pem = meta_get(&db, "tls_ca_key_pem").ok().flatten(); + let mut node_cert_pem = meta_get(&db, "tls_cert_pem").ok().flatten(); + let mut node_key_pem = meta_get(&db, "tls_key_pem").ok().flatten(); + + let join_tokens = Arc::new(RwLock::new(HashMap::::new())); + + if role == Role::Primary { + // Initialize CA if not present + if ca_cert_pem.is_none() || ca_key_pem.is_none() { + let (ca_cert, ca_key) = generate_ca().expect("Failed to generate cluster Root CA"); + meta_set(&db, "tls_ca_pem", &ca_cert).unwrap(); + meta_set(&db, "tls_ca_key_pem", &ca_key).unwrap(); + ca_cert_pem = Some(ca_cert); + ca_key_pem = Some(ca_key); + info!("Cluster Root CA initialized"); + } + + // Initialize Primary Server Certificate + if node_cert_pem.is_none() || node_key_pem.is_none() { + let (cert, key) = issue_leaf_cert( + ca_cert_pem.as_ref().unwrap(), + ca_key_pem.as_ref().unwrap(), + "primary", + vec!["localhost".into(), "127.0.0.1".into()], + ) + .expect("Failed to issue primary server certificate"); + meta_set(&db, "tls_cert_pem", &cert).unwrap(); + meta_set(&db, "tls_key_pem", &key).unwrap(); + node_cert_pem = Some(cert); + node_key_pem = Some(key); + info!("Primary server certificate generated"); + } + + // Preload an initial bootstrap join token if APP_JOIN_TOKEN is passed + if let Ok(seed_token) = std::env::var("APP_JOIN_TOKEN") { + join_tokens.write().unwrap().insert( + seed_token.clone(), + JoinToken { + token: seed_token, + expires_at: chrono::Utc::now() + chrono::Duration::hours(24), + used: false, + }, + ); + } + } else { + // Replica: Check if dynamic bootstrap enrollment is needed + if node_cert_pem.is_none() { + let p_url = primary_url + .as_ref() + .expect("APP_PRIMARY_URL required for replica enrollment"); + let token = std::env::var("APP_JOIN_TOKEN") + .expect("APP_JOIN_TOKEN required for initial replica enrollment"); + + info!(primary = %p_url, "Bootstrapping mTLS credentials via join token..."); + bootstrap_replica_join(p_url, &token, instance_id, &db) + .await + .expect("Dynamic join failed"); + + ca_cert_pem = meta_get(&db, "tls_ca_pem").ok().flatten(); + node_cert_pem = meta_get(&db, "tls_cert_pem").ok().flatten(); + node_key_pem = meta_get(&db, "tls_key_pem").ok().flatten(); + } + } let last_applied_seq = meta_get(&db, "last_applied_seq") .ok() @@ -998,14 +187,26 @@ async fn main() { last_success: None, last_error: None, })), + ca_cert_pem: ca_cert_pem.clone().map(Arc::new), + ca_key_pem: ca_key_pem.map(Arc::new), + join_tokens, }; if role == Role::Replica { + let sync_cfg = SyncConfig { + poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)), + max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)), + batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000), + }; + tokio::spawn(replica_loop( state.clone(), primary_url.expect("checked above"), - repl_token.expect("checked above"), + repl_token.unwrap_or_else(|| "".into()), sync_cfg, + ca_cert_pem.clone().unwrap(), + node_cert_pem.clone().unwrap(), + node_key_pem.clone().unwrap(), )); } @@ -1014,45 +215,45 @@ async fn main() { .route("/api/now", get(get_info)) .route("/api/records", get(list_records).post(create_record)) .route("/api/admin/config", get(get_config)) - .route( - "/api/admin/reload-config", - axum::routing::post(reload_config), - ) + .route("/api/admin/reload-config", post(reload_config)) .route("/api/replication/changes", get(replication_changes)) .route("/api/replication/status", get(replication_status)) - // gzip matters on slow links; reqwest's "gzip" feature negotiates it. + // Dynamic Peer Discovery Routes: + .route("/api/cluster/ca", get(get_cluster_ca)) + .route("/api/cluster/tokens", post(create_token_handler)) + .route("/api/cluster/join", post(join_cluster_handler)) .layer(tower_http::compression::CompressionLayer::new()) .with_state(state); - // Opt-in CORS for a browser dashboard, e.g. APP_CORS_ORIGIN=http://localhost:8080 - // (use "null" for a page opened from file://). GET only, single origin: a - // wildcard with POST would let any website you visit write records or - // trigger /api/admin/reload-config on your local instances. + let bind_str = std::env::var("APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into()); + let addr: SocketAddr = bind_str.parse().expect("Failed to parse APP_BIND address"); + + // Build mTLS ServerConfig + let server_config = create_mtls_server_config( + &ca_cert_pem.expect("CA cert missing"), + &node_cert_pem.expect("Server cert missing"), + &node_key_pem.expect("Server key missing"), + ) + .expect("Failed to build mTLS server configuration"); + let app = match std::env::var("APP_CORS_ORIGIN") { Ok(origin) => { info!(origin = %origin, "CORS enabled"); app.layer( tower_http::cors::CorsLayer::new() - .allow_origin( - origin - .parse::() - .expect("invalid APP_CORS_ORIGIN"), - ) + .allow_origin(origin.parse::().expect("invalid APP_CORS_ORIGIN")) .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) .allow_headers([CONTENT_TYPE, AUTHORIZATION]), ) } Err(_) => app, }; - - let addr = std::env::var("APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into()); - let listener = tokio::net::TcpListener::bind(&addr) - .await - .expect("Failed to bind address"); - info!(address = %addr, "HTTP server started"); + let rustls_config = RustlsConfig::from_config(Arc::new(server_config)); - axum::serve(listener, app) + info!(address = %addr, "mTLS HTTPS server listening"); + axum_server::bind_rustls(addr, rustls_config) + .serve(app.into_make_service()) .await - .expect("HTTP server failed"); + .expect("HTTPS server failed"); } diff --git a/src/replication.rs b/src/replication.rs new file mode 100644 index 0000000..ac5bd1e --- /dev/null +++ b/src/replication.rs @@ -0,0 +1,378 @@ +use axum::{ + extract::{Query, State}, + http::{header::AUTHORIZATION, HeaderMap, StatusCode}, + Json, +}; + +use crate::tls::create_mtls_client; +use chrono::Utc; +use rusqlite::{params, Connection}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use crate::{ + db::meta_set, + state::{api_error, db_error, internal_error, ApiError, AppState, SyncStatus}, +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct Change { + pub seq: i64, + pub uuid: Uuid, + pub date: String, + pub data: Value, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ChangesResponse { + pub instance_id: Uuid, + pub head_seq: i64, + pub has_more: bool, + pub changes: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct ChangesQuery { + pub after: Option, + pub limit: Option, +} + +#[derive(Debug, Clone)] +pub struct SyncConfig { + pub poll_interval: Duration, + pub max_backoff: Duration, + pub batch: i64, +} + +enum SyncError { + Transient(String), + Fatal(String), +} + +fn ct_eq(a: &[u8], b: &[u8]) -> bool { + a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 +} + +fn check_token(headers: &HeaderMap, expected: &Option>) -> Result<(), ApiError> { + let Some(expected) = expected else { + return Err(api_error( + StatusCode::FORBIDDEN, + "Replication is disabled (no token configured)", + )); + }; + + let provided = headers + .get(AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .unwrap_or(""); + + if !ct_eq(provided.as_bytes(), expected.as_bytes()) { + return Err(api_error(StatusCode::UNAUTHORIZED, "Invalid token")); + } + Ok(()) +} + +pub async fn replication_changes( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result, ApiError> { + check_token(&headers, &state.repl_token)?; + + let after = query.after.unwrap_or(0).max(0); + let limit = query.limit.unwrap_or(200).clamp(1, 1000); + + let db = state + .db + .lock() + .map_err(|_| internal_error("database lock poisoned"))?; + + let head_seq: i64 = db + .query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { + r.get(0) + }) + .map_err(|e| db_error("head seq", e))?; + + let mut statement = db + .prepare( + "SELECT seq, uuid, date, data + FROM records + WHERE seq > ?1 + ORDER BY seq ASC + LIMIT ?2", + ) + .map_err(|e| db_error("prepare changes", e))?; + + let rows = statement + .query_map(params![after, limit], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + )) + }) + .map_err(|e| db_error("query changes", e))?; + + let mut changes = Vec::new(); + for row in rows { + let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read change", e))?; + changes.push(Change { + seq, + uuid: Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?, + date, + data: serde_json::from_str(&data_json) + .map_err(|_| internal_error("invalid stored JSON"))?, + }); + } + + let last = changes.last().map(|c| c.seq).unwrap_or(after); + + Ok(Json(ChangesResponse { + instance_id: state.instance_id, + head_seq, + has_more: last < head_seq, + changes, + })) +} + +pub async fn replication_status(State(state): State) -> Result, ApiError> { + let status = state + .sync + .read() + .map_err(|_| internal_error("sync status lock poisoned"))? + .clone(); + + let local_head: i64 = { + let db = state + .db + .lock() + .map_err(|_| internal_error("database lock poisoned"))?; + db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { + r.get(0) + }) + .map_err(|e| db_error("head seq", e))? + }; + + let lag = status + .primary_head_seq + .map(|head| (head - status.last_applied_seq).max(0)); + + Ok(Json(serde_json::json!({ + "instance_id": state.instance_id, + "role": state.role.as_str(), + "primary_url": state.primary_url.as_deref(), + "local_head_seq": local_head, + "lag_records": lag, + "sync": status, + }))) +} + +fn jitter(d: Duration) -> Duration { + let ms = d.as_millis() as u64; + let extra = (Uuid::new_v4().as_u128() as u64) % (ms / 2 + 1); + Duration::from_millis(ms / 2 + extra) +} + +fn update_status(state: &AppState, f: impl FnOnce(&mut SyncStatus)) { + if let Ok(mut s) = state.sync.write() { + f(&mut s); + } +} + +fn apply_batch( + db: &Arc>, + source: Uuid, + changes: &[Change], +) -> Result { + let mut conn = db.lock().map_err(|_| "database lock poisoned".to_string())?; + let tx = conn.transaction().map_err(|e| e.to_string())?; + + let mut max_seq = 0; + { + let mut insert = tx + .prepare_cached( + "INSERT OR IGNORE INTO records (seq, uuid, date, data) VALUES (?1, ?2, ?3, ?4)", + ) + .map_err(|e| e.to_string())?; + + for c in changes { + let data = serde_json::to_string(&c.data).map_err(|e| e.to_string())?; + insert + .execute(params![c.seq, c.uuid.to_string(), c.date, data]) + .map_err(|e| e.to_string())?; + max_seq = max_seq.max(c.seq); + } + } + + meta_set(&tx, "last_applied_seq", &max_seq.to_string()).map_err(|e| e.to_string())?; + meta_set(&tx, "source_instance_id", &source.to_string()).map_err(|e| e.to_string())?; + + tx.commit().map_err(|e| e.to_string())?; + Ok(max_seq) +} + +async fn sync_once( + client: &reqwest::Client, + state: &AppState, + primary_url: &str, + token: &str, + batch: i64, +) -> Result { + let (last, known_source) = { + let s = state + .sync + .read() + .map_err(|_| SyncError::Fatal("sync status lock poisoned".into()))?; + (s.last_applied_seq, s.source_instance_id) + }; + + let url = format!("{}/api/replication/changes", primary_url.trim_end_matches('/')); + + let resp = client + .get(&url) + .bearer_auth(token) + .query(&[("after", last), ("limit", batch)]) + .send() + .await + .map_err(|e| SyncError::Transient(format!("request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(SyncError::Transient(format!( + "primary returned {}", + resp.status() + ))); + } + + let body: ChangesResponse = resp + .json() + .await + .map_err(|e| SyncError::Transient(format!("bad response body: {e}")))?; + + if body.instance_id == state.instance_id { + return Err(SyncError::Fatal( + "source has the same instance_id as this replica (cloned database?)".into(), + )); + } + if let Some(known) = known_source { + if known != body.instance_id { + return Err(SyncError::Fatal(format!( + "source instance changed: expected {known}, got {}", + body.instance_id + ))); + } + } + if body.head_seq < last { + return Err(SyncError::Fatal(format!( + "source head_seq {} is behind our cursor {last} (source restored from older backup?)", + body.head_seq + ))); + } + + if body.changes.is_empty() { + update_status(state, |s| { + s.state = "idle".into(); + s.primary_head_seq = Some(body.head_seq); + s.last_success = Some(Utc::now().to_rfc3339()); + s.last_error = None; + s.source_instance_id = Some(body.instance_id); + }); + return Ok(false); + } + + let mut prev = last; + for c in &body.changes { + if c.seq <= prev { + return Err(SyncError::Transient(format!( + "non-monotonic seq {} after {prev}", + c.seq + ))); + } + prev = c.seq; + } + + let count = body.changes.len(); + let db = state.db.clone(); + let source = body.instance_id; + let changes = body.changes; + + let max_seq = tokio::task::spawn_blocking(move || apply_batch(&db, source, &changes)) + .await + .map_err(|e| SyncError::Transient(format!("apply task failed: {e}")))? + .map_err(|e| SyncError::Transient(format!("apply failed: {e}")))?; + + update_status(state, |s| { + s.state = if body.has_more { "syncing" } else { "idle" }.into(); + s.last_applied_seq = max_seq; + s.source_instance_id = Some(source); + s.primary_head_seq = Some(body.head_seq); + s.last_success = Some(Utc::now().to_rfc3339()); + s.last_error = None; + }); + + info!( + applied = count, + last_applied_seq = max_seq, + head = body.head_seq, + "Replication batch applied" + ); + + Ok(body.has_more) +} + +pub async fn replica_loop( + state: AppState, + primary_url: String, + token: String, + cfg: SyncConfig, + ca_pem: String, + cert_pem: String, + key_pem: String, +) { + let client = match create_mtls_client(&ca_pem, &cert_pem, &key_pem) { + Ok(c) => c, + Err(e) => { + error!(error = %e, "Cannot build mTLS HTTP client; replication halted"); + return; + } + }; + + let mut backoff = Duration::from_secs(1); + info!(primary = %primary_url, "Replica mTLS sync loop started"); + + loop { + match sync_once(&client, &state, &primary_url, &token, cfg.batch).await { + Ok(more) => { + backoff = Duration::from_secs(1); + if !more { + tokio::time::sleep(jitter(cfg.poll_interval)).await; + } + } + Err(SyncError::Transient(message)) => { + let wait = jitter(backoff); + warn!(reason = %message, retry_in = ?wait, "Replication pull failed"); + update_status(&state, |s| { + s.state = "retrying".into(); + s.last_error = Some(message); + }); + tokio::time::sleep(wait).await; + backoff = (backoff * 2).min(cfg.max_backoff); + } + Err(SyncError::Fatal(message)) => { + error!(reason = %message, "Replication halted; operator action required"); + update_status(&state, |s| { + s.state = "halted".into(); + s.last_error = Some(message); + }); + return; + } + } + } +} diff --git a/src/schema.rs b/src/schema.rs new file mode 100644 index 0000000..1870bbc --- /dev/null +++ b/src/schema.rs @@ -0,0 +1,105 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::{collections::HashMap, fs}; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ValidationMode { + Strict, + Loose, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FieldSpec { + #[serde(rename = "type")] + pub data_type: DataType, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum DataType { + String, + Integer, + Number, + Boolean, + Object, + Array, + Null, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SchemaConfig { + pub mode: ValidationMode, + pub fields: HashMap, +} + +pub fn matches_type(value: &Value, expected: &DataType) -> bool { + match expected { + DataType::String => value.is_string(), + DataType::Integer => value.as_i64().is_some() || value.as_u64().is_some(), + DataType::Number => value.is_number(), + DataType::Boolean => value.is_boolean(), + DataType::Object => value.is_object(), + DataType::Array => value.is_array(), + DataType::Null => value.is_null(), + } +} + +pub fn type_name(data_type: &DataType) -> &'static str { + match data_type { + DataType::String => "string", + DataType::Integer => "integer", + DataType::Number => "number", + DataType::Boolean => "boolean", + DataType::Object => "object", + DataType::Array => "array", + DataType::Null => "null", + } +} + +pub fn validate_data(data: &Value, schema: &SchemaConfig) -> Result<(), String> { + let object = data.as_object().ok_or("data must be a JSON object")?; + + for (key, spec) in &schema.fields { + let value = object + .get(key) + .ok_or_else(|| format!("Missing required field: {key}"))?; + + if !matches_type(value, &spec.data_type) { + return Err(format!( + "Field '{key}' must be of type {}", + type_name(&spec.data_type) + )); + } + + if let Some(expected) = &spec.value { + if value != expected { + return Err(format!("Field '{key}' must equal {expected}")); + } + } + } + + if matches!(schema.mode, ValidationMode::Strict) { + for key in object.keys() { + if !schema.fields.contains_key(key) { + return Err(format!("Unknown field: {key}")); + } + } + } + + Ok(()) +} + +pub fn load_config(path: &str) -> Result { + let contents = fs::read_to_string(path).map_err(|e| format!("Cannot read {path}: {e}"))?; + let config: SchemaConfig = + serde_json::from_str(&contents).map_err(|e| format!("Invalid configuration JSON: {e}"))?; + + if config.fields.is_empty() { + return Err("Configuration must define at least one field".into()); + } + + Ok(config) +} \ No newline at end of file diff --git a/src/state.rs b/src/state.rs new file mode 100644 index 0000000..19a680d --- /dev/null +++ b/src/state.rs @@ -0,0 +1,84 @@ +use axum::http::StatusCode; +use rusqlite::Connection; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::{ + collections::HashMap, + sync::{Arc, Mutex, RwLock}, +}; +use tracing::error; +use uuid::Uuid; + +use crate::{schema::SchemaConfig, tls::JoinToken}; + +#[derive(Clone)] +pub struct AppState { + pub db: Arc>, + pub schema: Arc>, + pub config_path: Arc, + pub instance_id: Uuid, + pub role: Role, + pub primary_url: Option>, + pub repl_token: Option>, + pub sync: Arc>, + + // mTLS & Discovery additions: + pub ca_cert_pem: Option>, + pub ca_key_pem: Option>, + pub join_tokens: Arc>>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Role { + Primary, + Replica, +} + +impl Role { + pub fn as_str(self) -> &'static str { + match self { + Role::Primary => "primary", + Role::Replica => "replica", + } + } +} + +#[derive(Debug, Clone, Serialize)] +pub struct SyncStatus { + pub state: String, + pub last_applied_seq: i64, + pub source_instance_id: Option, + pub primary_head_seq: Option, + pub last_success: Option, + pub last_error: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ApiResponse { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub seq: Option, + pub uuid: Uuid, + pub date: String, + pub data: Value, +} + +#[derive(Debug, Deserialize)] +pub struct CreateRequest { + pub data: Value, +} + +pub type ApiError = (StatusCode, String); + +pub fn api_error(status: StatusCode, message: impl Into) -> ApiError { + (status, message.into()) +} + +pub fn internal_error(context: &str) -> ApiError { + error!(context, "Internal server error"); + api_error(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error") +} + +pub fn db_error(context: &str, e: rusqlite::Error) -> ApiError { + error!(error = %e, context, "Database error"); + api_error(StatusCode::INTERNAL_SERVER_ERROR, "Database error") +} diff --git a/src/tls.rs b/src/tls.rs new file mode 100644 index 0000000..f48fc9d --- /dev/null +++ b/src/tls.rs @@ -0,0 +1,219 @@ +use std::sync::Arc; +use chrono::{DateTime, Duration as ChronoDuration, Utc}; +use rcgen::{ + BasicConstraints, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa, + Issuer, KeyPair, KeyUsagePurpose, +}; +use rusqlite::Connection; +use rustls::server::WebPkiClientVerifier; +use rustls::RootCertStore; +use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use serde::{Deserialize, Serialize}; +use tracing::info; +use uuid::Uuid; + +use crate::db::{meta_set}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JoinToken { + pub token: String, + pub expires_at: DateTime, + pub used: bool, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct JoinClusterRequest { + pub token: String, + pub node_id: Uuid, + #[serde(default)] + pub san_list: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct JoinClusterResponse { + pub ca_pem: String, + pub cert_pem: String, + pub key_pem: String, +} + +// --------------------------------------------------------------------------- +// Certificate Generation +// --------------------------------------------------------------------------- + +pub fn generate_ca() -> Result<(String, String), String> { + let ca_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256) + .map_err(|e| format!("CA key error: {e}"))?; + + let mut ca_params = CertificateParams::new(vec![]) + .map_err(|e| format!("CA params error: {e}"))?; + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + ca_params.distinguished_name.push(DnType::CommonName, "Cluster Root CA"); + ca_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign]; + + let ca_cert = ca_params + .self_signed(&ca_key) + .map_err(|e| format!("CA self-sign error: {e}"))?; + + Ok((ca_cert.pem(), ca_key.serialize_pem())) +} + +pub fn issue_leaf_cert( + ca_cert_pem: &str, + ca_key_pem: &str, + common_name: &str, + san_list: Vec, +) -> Result<(String, String), String> { + let ca_key = KeyPair::from_pem(ca_key_pem) + .map_err(|e| format!("Failed to parse CA private key: {e}"))?; + + // In rcgen 0.14+, from_ca_cert_pem belongs to Issuer: + let issuer = Issuer::from_ca_cert_pem(ca_cert_pem, &ca_key) + .map_err(|e| format!("Failed to parse CA cert/create issuer: {e}"))?; + + let leaf_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256) + .map_err(|e| format!("Leaf key error: {e}"))?; + + let mut leaf_params = CertificateParams::new(san_list) + .map_err(|e| format!("Leaf params error: {e}"))?; + leaf_params.distinguished_name.push(DnType::CommonName, common_name); + leaf_params.use_authority_key_identifier_extension = true; + leaf_params.key_usages = vec![KeyUsagePurpose::DigitalSignature]; + leaf_params.extended_key_usages = vec![ + ExtendedKeyUsagePurpose::ServerAuth, + ExtendedKeyUsagePurpose::ClientAuth, + ]; + + let leaf_cert = leaf_params + .signed_by(&leaf_key, &issuer) + .map_err(|e| format!("Leaf sign error: {e}"))?; + + Ok((leaf_cert.pem(), leaf_key.serialize_pem())) +} + +// --------------------------------------------------------------------------- +// Rustls Server & Reqwest Client Configuration +// --------------------------------------------------------------------------- + +fn parse_certs(pem: &str) -> Result>, String> { + let mut reader = pem.as_bytes(); + let certs: Result, _> = rustls_pemfile::certs(&mut reader).collect(); + certs.map_err(|e| format!("failed to parse certs: {e}")) +} + +fn parse_key(pem: &str) -> Result, String> { + let mut reader = pem.as_bytes(); + rustls_pemfile::private_key(&mut reader) + .map_err(|e| format!("failed to parse private key: {e}"))? + .ok_or_else(|| "no private key found in pem".to_string()) +} + +pub fn create_mtls_server_config( + ca_pem: &str, + server_cert_pem: &str, + server_key_pem: &str, +) -> Result { + let mut root_store = RootCertStore::empty(); + for cert in parse_certs(ca_pem)? { + root_store + .add(cert) + .map_err(|e| format!("failed to add root cert: {e}"))?; + } + + // allow_unauthenticated permits anonymous TLS connections (e.g., initial join requests) + // while enforcing valid CA signatures on any presented client certificates. + let client_verifier = WebPkiClientVerifier::builder(Arc::new(root_store)) + .allow_unauthenticated() + .build() + .map_err(|e| format!("failed to build client verifier: {e}"))?; + + let cert_chain = parse_certs(server_cert_pem)?; + let key = parse_key(server_key_pem)?; + + let mut config = rustls::ServerConfig::builder() + .with_client_cert_verifier(client_verifier) + .with_single_cert(cert_chain, key) + .map_err(|e| format!("failed to build ServerConfig: {e}"))?; + + config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; + Ok(config) +} + +pub fn create_mtls_client( + ca_pem: &str, + client_cert_pem: &str, + client_key_pem: &str, +) -> Result { + let root_cert = reqwest::Certificate::from_pem(ca_pem.as_bytes()) + .map_err(|e| format!("failed to load CA pem for client: {e}"))?; + + let identity_pem = format!("{client_cert_pem}\n{client_key_pem}"); + let identity = reqwest::Identity::from_pem(identity_pem.as_bytes()) + .map_err(|e| format!("failed to create client identity: {e}"))?; + + reqwest::Client::builder() + .use_rustls_tls() + .add_root_certificate(root_cert) + .identity(identity) + .connect_timeout(std::time::Duration::from_secs(10)) + .timeout(std::time::Duration::from_secs(60)) + .build() + .map_err(|e| format!("failed to build reqwest client: {e}")) +} + +// --------------------------------------------------------------------------- +// Dynamic Enrollment / Join Helpers +// --------------------------------------------------------------------------- + +pub fn create_join_token(ttl_secs: u64) -> JoinToken { + JoinToken { + token: Uuid::new_v4().to_string().replace('-', ""), + expires_at: Utc::now() + ChronoDuration::seconds(ttl_secs as i64), + used: false, + } +} + +pub async fn bootstrap_replica_join( + primary_url: &str, + token: &str, + node_id: Uuid, + db: &Connection, +) -> Result<(), String> { + let base = primary_url.trim_end_matches('/'); + + // 1. Initial bootstrap client without mTLS client cert + let bootstrap_client = reqwest::Client::builder() + .danger_accept_invalid_certs(true) + .build() + .map_err(|e| format!("bootstrap client build failed: {e}"))?; + + // 2. Request enrollment + let join_req = JoinClusterRequest { + token: token.to_string(), + node_id, + san_list: vec!["localhost".into(), "127.0.0.1".into()], + }; + + let resp = bootstrap_client + .post(format!("{base}/api/cluster/join")) + .json(&join_req) + .send() + .await + .map_err(|e| format!("join request failed: {e}"))?; + + if !resp.status().is_success() { + return Err(format!("join rejected with HTTP {}", resp.status())); + } + + let payload: JoinClusterResponse = resp + .json() + .await + .map_err(|e| format!("failed to decode join response: {e}"))?; + + // 3. Persist credentials locally + meta_set(db, "tls_ca_pem", &payload.ca_pem).map_err(|e| e.to_string())?; + meta_set(db, "tls_cert_pem", &payload.cert_pem).map_err(|e| e.to_string())?; + meta_set(db, "tls_key_pem", &payload.key_pem).map_err(|e| e.to_string())?; + + info!("Replica successfully joined cluster and persisted mTLS certificates"); + Ok(()) +}