diff --git a/Cargo.lock b/Cargo.lock index 02c0cad..84f18b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,17 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "ahash" -version = "0.7.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "891477e0c6a8957309ee5c45a6368af3ae14bb510732d2684ffa19af310920f9" -dependencies = [ - "getrandom 0.2.17", - "once_cell", - "version_check", -] - [[package]] name = "aho-corasick" version = "1.1.4" @@ -22,39 +11,12 @@ dependencies = [ "memchr", ] -[[package]] -name = "allocator-api2" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" - -[[package]] -name = "android_system_properties" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" -dependencies = [ - "libc", -] - [[package]] name = "anyhow" version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" -[[package]] -name = "array-init" -version = "2.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc" - -[[package]] -name = "arrayvec" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" - [[package]] name = "async-trait" version = "0.1.89" @@ -63,15 +25,9 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] -[[package]] -name = "autocfg" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" - [[package]] name = "aws-lc-rs" version = "1.16.2" @@ -79,7 +35,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc" dependencies = [ "aws-lc-sys", - "untrusted 0.7.1", + "untrusted", "zeroize", ] @@ -101,22 +57,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "base64ct" -version = "1.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" - -[[package]] -name = "bcder" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f7c42c9913f68cf9390a225e81ad56a5c515347287eb98baa710090ca1de86d" -dependencies = [ - "bytes", - "smallvec", -] - [[package]] name = "bitflags" version = "2.11.0" @@ -124,15 +64,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" [[package]] -name = "bitvec" -version = "1.0.1" +name = "block-buffer" +version = "0.10.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bc2832c24239b0141d5674bb9174f9d68a8b5b3f2753311927c172ca46f7e9c" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" dependencies = [ - "funty", - "radium", - "tap", - "wyz", + "generic-array", ] [[package]] @@ -144,58 +81,12 @@ dependencies = [ "hybrid-array", ] -[[package]] -name = "borsh" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfd1e3f8955a5d7de9fab72fc8373fade9fb8a703968cb200ae3dc6cf08e185a" -dependencies = [ - "borsh-derive", - "bytes", - "cfg_aliases", -] - -[[package]] -name = "borsh-derive" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfcfdc083699101d5a7965e49925975f2f55060f94f9a05e7187be95d530ca59" -dependencies = [ - "once_cell", - "proc-macro-crate", - "proc-macro2", - "quote", - "syn 2.0.117", -] - [[package]] name = "bumpalo" version = "3.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" -[[package]] -name = "bytecheck" -version = "0.6.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23cdc57ce23ac53c931e88a43d06d070a6fd142f2617be5855eb75efc9beb1c2" -dependencies = [ - "bytecheck_derive", - "ptr_meta", - "simdutf8", -] - -[[package]] -name = "bytecheck_derive" -version = "0.6.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3db406d29fbcd95542e92559bed4d8ad92636d1ca8b3b72ede10b4bcc010e659" -dependencies = [ - "proc-macro2", - "quote", - "syn 1.0.109", -] - [[package]] name = "byteorder" version = "1.5.0" @@ -226,12 +117,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" -[[package]] -name = "cfg_aliases" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" - [[package]] name = "chacha20" version = "0.10.0" @@ -239,23 +124,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.0", "rand_core 0.10.0", ] -[[package]] -name = "chrono" -version = "0.4.44" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" -dependencies = [ - "iana-time-zone", - "js-sys", - "num-traits", - "wasm-bindgen", - "windows-link", -] - [[package]] name = "cmake" version = "0.1.58" @@ -271,12 +143,6 @@ version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f88a43d011fc4a6876cb7344703e297c71dda42494fee094d5f7c76bf13f746" -[[package]] -name = "const-oid" -version = "0.9.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" - [[package]] name = "const-oid" version = "0.10.2" @@ -284,10 +150,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" [[package]] -name = "core-foundation-sys" -version = "0.8.7" +name = "cpufeatures" +version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] [[package]] name = "cpufeatures" @@ -298,6 +167,16 @@ dependencies = [ "libc", ] +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "crypto-common" version = "0.2.1" @@ -317,59 +196,14 @@ dependencies = [ ] [[package]] -name = "deadpool" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" -dependencies = [ - "deadpool-runtime", - "lazy_static", - "num_cpus", - "tokio", -] - -[[package]] -name = "deadpool-postgres" -version = "0.14.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d697d376cbfa018c23eb4caab1fd1883dd9c906a8c034e8d9a3cb06a7e0bef9" -dependencies = [ - "async-trait", - "deadpool", - "getrandom 0.2.17", - "tokio", - "tokio-postgres", - "tracing", -] - -[[package]] -name = "deadpool-runtime" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" -dependencies = [ - "tokio", -] - -[[package]] -name = "der" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" -dependencies = [ - "const-oid 0.9.6", - "zeroize", -] - -[[package]] -name = "derive-new" -version = "0.7.0" +name = "digest" +version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cdc8d50f426189eef89dac62fabfa0abb27d5cc008f25bf4156a0203325becc" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", + "block-buffer 0.10.4", + "crypto-common 0.1.7", + "subtle", ] [[package]] @@ -378,9 +212,9 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4850db49bf08e663084f7fb5c87d202ef91a3907271aff24a94eb97ff039153c" dependencies = [ - "block-buffer", - "const-oid 0.10.2", - "crypto-common", + "block-buffer 0.12.0", + "const-oid", + "crypto-common 0.2.1", "ctutils", ] @@ -396,6 +230,16 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "fallible-iterator" version = "0.2.0" @@ -414,39 +258,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" -[[package]] -name = "foldhash" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" - [[package]] name = "fs_extra" version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" -[[package]] -name = "funty" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" - -[[package]] -name = "futures" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" -dependencies = [ - "futures-channel", - "futures-core", - "futures-executor", - "futures-io", - "futures-sink", - "futures-task", - "futures-util", -] - [[package]] name = "futures-channel" version = "0.3.32" @@ -463,34 +280,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" -[[package]] -name = "futures-executor" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" -dependencies = [ - "futures-core", - "futures-task", - "futures-util", -] - -[[package]] -name = "futures-io" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" - -[[package]] -name = "futures-macro" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", -] - [[package]] name = "futures-sink" version = "0.3.32" @@ -509,15 +298,20 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ - "futures-channel", "futures-core", - "futures-io", - "futures-macro", "futures-sink", "futures-task", - "memchr", "pin-project-lite", - "slab", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", ] [[package]] @@ -527,10 +321,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", - "js-sys", "libc", "wasi 0.11.1+wasi-snapshot-preview1", - "wasm-bindgen", ] [[package]] @@ -559,22 +351,13 @@ dependencies = [ "wasip3", ] -[[package]] -name = "hashbrown" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" -dependencies = [ - "ahash", -] - [[package]] name = "hashbrown" version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "foldhash 0.1.5", + "foldhash", ] [[package]] @@ -582,11 +365,6 @@ name = "hashbrown" version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" -dependencies = [ - "allocator-api2", - "equivalent", - "foldhash 0.2.0", -] [[package]] name = "heck" @@ -595,16 +373,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] -name = "hermit-abi" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" - -[[package]] -name = "hex" -version = "0.4.3" +name = "hmac" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest 0.10.7", +] [[package]] name = "hmac" @@ -612,7 +387,7 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" dependencies = [ - "digest", + "digest 0.11.2", ] [[package]] @@ -624,30 +399,6 @@ dependencies = [ "typenum", ] -[[package]] -name = "iana-time-zone" -version = "0.1.65" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" -dependencies = [ - "android_system_properties", - "core-foundation-sys", - "iana-time-zone-haiku", - "js-sys", - "log", - "wasm-bindgen", - "windows-core", -] - -[[package]] -name = "iana-time-zone-haiku" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" -dependencies = [ - "cc", -] - [[package]] name = "id-arena" version = "2.3.0" @@ -707,29 +458,6 @@ dependencies = [ "signature", ] -[[package]] -name = "lazy-regex" -version = "3.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6bae91019476d3ec7147de9aa291cadb6d870abf2f3015d2da73a90325ac1496" -dependencies = [ - "lazy-regex-proc_macros", - "once_cell", - "regex-lite", -] - -[[package]] -name = "lazy-regex-proc_macros" -version = "3.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4de9c1e1439d8b7b3061b2d209809f447ca33241733d9a3c01eabf2dc8d94358" -dependencies = [ - "proc-macro2", - "quote", - "regex", - "syn 2.0.117", -] - [[package]] name = "lazy_static" version = "1.5.0" @@ -772,15 +500,6 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" -[[package]] -name = "lru" -version = "0.16.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dc47f592c06f33f8e3aea9591776ec7c9f9e4124778ff8a3c3b87159f7e593" -dependencies = [ - "hashbrown 0.16.1", -] - [[package]] name = "matchers" version = "0.2.0" @@ -797,15 +516,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" dependencies = [ "cfg-if", - "digest", + "digest 0.11.2", ] -[[package]] -name = "md5" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae960838283323069879657ca3de837e9f7bbb4c7bf6ea7f1b290d5e9476d2e0" - [[package]] name = "memchr" version = "2.8.0" @@ -820,7 +533,7 @@ checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi 0.11.1+wasi-snapshot-preview1", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -829,26 +542,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", -] - -[[package]] -name = "num-traits" -version = "0.2.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" -dependencies = [ - "autocfg", -] - -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", + "windows-sys", ] [[package]] @@ -898,78 +592,24 @@ dependencies = [ "windows-link", ] -[[package]] -name = "pem" -version = "3.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" -dependencies = [ - "base64", - "serde_core", -] - [[package]] name = "percent-encoding" version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" -[[package]] -name = "pg_interval_2" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a055f44628dcf9c4e68f931535dabd3544a239655fdde25a3b0e95d4b36e9260" -dependencies = [ - "bytes", - "chrono", - "postgres-types", -] - -[[package]] -name = "pgwire" -version = "0.38.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a1bdf05fc8231cc5024572fe056e3ce34eb6b9b755ba7aba110e1c64119cec3" -dependencies = [ - "async-trait", - "aws-lc-rs", - "base64", - "bytes", - "chrono", - "derive-new", - "futures", - "hex", - "lazy-regex", - "md5", - "pg_interval_2", - "postgres-types", - "rand 0.10.0", - "rust_decimal", - "rustls-pki-types", - "ryu", - "serde", - "serde_json", - "smol_str", - "stringprep", - "thiserror", - "tokio", - "tokio-rustls", - "tokio-util", - "x509-certificate", -] - [[package]] name = "pgwire-supabase-proxy" version = "0.1.0" dependencies = [ - "async-trait", + "base64", "bytes", - "deadpool-postgres", - "futures", + "hmac 0.12.1", "jsonwebtoken", - "lru", - "pgwire", + "rand 0.8.5", "serde", + "serde_json", + "sha2 0.10.9", "thiserror", "tokio", "tokio-postgres", @@ -1013,11 +653,11 @@ dependencies = [ "byteorder", "bytes", "fallible-iterator", - "hmac", + "hmac 0.13.0", "md-5", "memchr", "rand 0.10.0", - "sha2", + "sha2 0.11.0", "stringprep", ] @@ -1027,13 +667,9 @@ version = "0.2.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8dc729a129e682e8d24170cd30ae1aa01b336b096cbb56df6d534ffec133d186" dependencies = [ - "array-init", "bytes", - "chrono", "fallible-iterator", "postgres-protocol", - "serde_core", - "serde_json", ] [[package]] @@ -1052,16 +688,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn 2.0.117", -] - -[[package]] -name = "proc-macro-crate" -version = "3.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f" -dependencies = [ - "toml_edit", + "syn", ] [[package]] @@ -1073,26 +700,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "ptr_meta" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0738ccf7ea06b608c10564b31debd4f5bc5e197fc8bfe088f68ae5ce81e7a4f1" -dependencies = [ - "ptr_meta_derive", -] - -[[package]] -name = "ptr_meta_derive" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "16b845dbfca988fa33db069c0e230574d15a3088f147a87b64c7589eb662c9ac" -dependencies = [ - "proc-macro2", - "quote", - "syn 1.0.109", -] - [[package]] name = "quote" version = "1.0.45" @@ -1112,13 +719,7 @@ checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" name = "r-efi" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" - -[[package]] -name = "radium" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" [[package]] name = "rand" @@ -1176,18 +777,6 @@ dependencies = [ "bitflags", ] -[[package]] -name = "regex" -version = "1.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" -dependencies = [ - "aho-corasick", - "memchr", - "regex-automata", - "regex-syntax", -] - [[package]] name = "regex-automata" version = "0.4.14" @@ -1199,153 +788,29 @@ dependencies = [ "regex-syntax", ] -[[package]] -name = "regex-lite" -version = "0.1.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" - [[package]] name = "regex-syntax" version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" -[[package]] -name = "rend" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71fe3824f5629716b1589be05dacd749f6aa084c87e00e016714a8cdfccc997c" -dependencies = [ - "bytecheck", -] - -[[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 0.9.0", - "windows-sys 0.52.0", -] - -[[package]] -name = "rkyv" -version = "0.7.46" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2297bf9c81a3f0dc96bc9521370b88f054168c29826a75e89c55ff196e7ed6a1" -dependencies = [ - "bitvec", - "bytecheck", - "bytes", - "hashbrown 0.12.3", - "ptr_meta", - "rend", - "rkyv_derive", - "seahash", - "tinyvec", - "uuid", -] - -[[package]] -name = "rkyv_derive" -version = "0.7.46" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "84d7b42d4b8d06048d3ac8db0eb31bcb942cbeb709f0b5f2b2ebde398d3038f5" -dependencies = [ - "proc-macro2", - "quote", - "syn 1.0.109", -] - -[[package]] -name = "rust_decimal" -version = "1.41.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ce901f9a19d251159075a4c37af514c3b8ef99c22e02dd8c19161cf397ee94a" -dependencies = [ - "arrayvec", - "borsh", - "bytes", - "num-traits", - "postgres-types", - "rand 0.8.5", - "rkyv", - "serde", - "serde_json", - "wasm-bindgen", -] - -[[package]] -name = "rustls" -version = "0.23.37" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" -dependencies = [ - "aws-lc-rs", - "log", - "once_cell", - "rustls-pki-types", - "rustls-webpki", - "subtle", - "zeroize", -] - -[[package]] -name = "rustls-pki-types" -version = "1.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd" -dependencies = [ - "zeroize", -] - -[[package]] -name = "rustls-webpki" -version = "0.103.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" -dependencies = [ - "aws-lc-rs", - "ring", - "rustls-pki-types", - "untrusted 0.9.0", -] - [[package]] name = "rustversion" version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" -[[package]] -name = "ryu" -version = "1.0.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" - [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "seahash" -version = "4.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" - [[package]] name = "semver" -version = "1.0.27" +version = "1.0.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" [[package]] name = "serde" @@ -1374,7 +839,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] [[package]] @@ -1390,6 +855,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha2" version = "0.11.0" @@ -1397,8 +873,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.3.0", + "digest 0.11.2", ] [[package]] @@ -1416,6 +892,16 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "signature" version = "2.2.0" @@ -1425,40 +911,18 @@ dependencies = [ "rand_core 0.6.4", ] -[[package]] -name = "simdutf8" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" - [[package]] name = "siphasher" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e" -[[package]] -name = "slab" -version = "0.4.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" - [[package]] name = "smallvec" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" -[[package]] -name = "smol_str" -version = "0.3.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4aaa7368fcf4852a4c2dd92df0cace6a71f2091ca0a23391ce7f3a31833f1523" -dependencies = [ - "borsh", - "serde_core", -] - [[package]] name = "socket2" version = "0.6.3" @@ -1466,17 +930,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", -] - -[[package]] -name = "spki" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" -dependencies = [ - "base64ct", - "der", + "windows-sys", ] [[package]] @@ -1496,17 +950,6 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" -[[package]] -name = "syn" -version = "1.0.109" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" -dependencies = [ - "proc-macro2", - "quote", - "unicode-ident", -] - [[package]] name = "syn" version = "2.0.117" @@ -1518,12 +961,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "tap" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" - [[package]] name = "thiserror" version = "2.0.18" @@ -1541,7 +978,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] [[package]] @@ -1578,9 +1015,10 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -1591,7 +1029,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] [[package]] @@ -1620,16 +1058,6 @@ dependencies = [ "whoami", ] -[[package]] -name = "tokio-rustls" -version = "0.26.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" -dependencies = [ - "rustls", - "tokio", -] - [[package]] name = "tokio-stream" version = "0.1.18" @@ -1665,36 +1093,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "toml_datetime" -version = "1.1.1+spec-1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" -dependencies = [ - "serde_core", -] - -[[package]] -name = "toml_edit" -version = "0.25.10+spec-1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a82418ca169e235e6c399a84e395ab6debeb3bc90edc959bf0f48647c6a32d1b" -dependencies = [ - "indexmap", - "toml_datetime", - "toml_parser", - "winnow", -] - -[[package]] -name = "toml_parser" -version = "1.1.2+spec-1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" -dependencies = [ - "winnow", -] - [[package]] name = "tracing" version = "0.1.44" @@ -1714,7 +1112,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] [[package]] @@ -1801,22 +1199,6 @@ version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" -[[package]] -name = "untrusted" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" - -[[package]] -name = "uuid" -version = "1.23.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ac8b6f42ead25368cf5b098aeb3dc8a1a2c05a3eee8a9a1a68c640edbfc79d9" -dependencies = [ - "js-sys", - "wasm-bindgen", -] - [[package]] name = "valuable" version = "0.1.1" @@ -1880,7 +1262,6 @@ dependencies = [ "cfg-if", "once_cell", "rustversion", - "serde", "wasm-bindgen-macro", "wasm-bindgen-shared", ] @@ -1904,7 +1285,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 2.0.117", + "syn", "wasm-bindgen-shared", ] @@ -1974,74 +1355,12 @@ dependencies = [ "web-sys", ] -[[package]] -name = "windows-core" -version = "0.62.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" -dependencies = [ - "windows-implement", - "windows-interface", - "windows-link", - "windows-result", - "windows-strings", -] - -[[package]] -name = "windows-implement" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", -] - -[[package]] -name = "windows-interface" -version = "0.59.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", -] - [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-result" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" -dependencies = [ - "windows-link", -] - -[[package]] -name = "windows-strings" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" -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" @@ -2051,79 +1370,6 @@ 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 = "winnow" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09dac053f1cd375980747450bfc7250c264eaae0583872e845c0c7cd578872b5" -dependencies = [ - "memchr", -] - [[package]] name = "wit-bindgen" version = "0.51.0" @@ -2154,7 +1400,7 @@ dependencies = [ "heck", "indexmap", "prettyplease", - "syn 2.0.117", + "syn", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -2170,7 +1416,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn 2.0.117", + "syn", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -2212,34 +1458,6 @@ dependencies = [ "wasmparser", ] -[[package]] -name = "wyz" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f360fc0b24296329c78fda852a1e9ae82de9cf7b27dae4b7f62f118f77b9ed" -dependencies = [ - "tap", -] - -[[package]] -name = "x509-certificate" -version = "0.25.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca9eb9a0c822c67129d5b8fcc2806c6bc4f50496b420825069a440669bcfbf7f" -dependencies = [ - "bcder", - "bytes", - "chrono", - "der", - "hex", - "pem", - "ring", - "signature", - "spki", - "thiserror", - "zeroize", -] - [[package]] name = "zerocopy" version = "0.8.48" @@ -2257,7 +1475,7 @@ checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn", ] [[package]] @@ -2265,20 +1483,6 @@ name = "zeroize" version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" -dependencies = [ - "zeroize_derive", -] - -[[package]] -name = "zeroize_derive" -version = "1.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", -] [[package]] name = "zmij" diff --git a/Cargo.toml b/Cargo.toml index 7a9fd67..5ab69f5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,20 +3,51 @@ name = "pgwire-supabase-proxy" version = "0.1.0" edition = "2021" +[lib] +name = "pgwire_supabase_proxy" +path = "src/lib.rs" + +[[bin]] +name = "pgwire-supabase-proxy" +path = "src/main.rs" + [dependencies] -pgwire = { version = "0.38", features = ["server-api-aws-lc-rs"] } -tokio-postgres = "0.7" -deadpool-postgres = "0.14" -tokio = { version = "1", features = ["rt-multi-thread", "net", "macros", "sync"] } -async-trait = "0.1" -futures = "0.3" +# Async runtime +tokio = { version = "1", features = ["rt-multi-thread", "net", "macros", "sync", "signal", "io-util"] } + +# Wire protocol bytes = "1" + +# Auth jsonwebtoken = { version = "10", default-features = false, features = ["aws_lc_rs"] } + +# Serialization serde = { version = "1", features = ["derive"] } +serde_json = "1" + +# Error handling thiserror = "2" + +# Logging tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } -lru = "0.16" + +# SCRAM +hmac = "0.12" +sha2 = "0.10" + +# Base64 (SCRAM) +base64 = "0.22" + +# Random +rand = "0.8" + +# TLS (optional — enable with features = ["tls"]; currently disabled for no-TLS MVP) +# tokio-rustls = "0.26" +# rustls = "0.23" +# webpki-roots = "0.26" [dev-dependencies] tokio-test = "0.4" +tokio = { version = "1", features = ["rt-multi-thread", "process", "fs"] } +tokio-postgres = "0.7" diff --git a/scripts/run-integration-tests.sh b/scripts/run-integration-tests.sh new file mode 100755 index 0000000..d92786c --- /dev/null +++ b/scripts/run-integration-tests.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# Run integration tests against orbstack postgres-dev. +# Manages kubectl port-forward lifecycle: start → test → cleanup. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(dirname "$SCRIPT_DIR")" +LOCAL_PORT=5433 +NAMESPACE=default +SERVICE=postgres-dev-postgres-dev + +cleanup() { + if [[ -n "${PF_PID:-}" ]] && kill -0 "$PF_PID" 2>/dev/null; then + echo "killing kubectl port-forward (PID $PF_PID)..." + kill "$PF_PID" 2>/dev/null || true + wait "$PF_PID" 2>/dev/null || true + fi +} +trap cleanup EXIT + +# Kill any existing port-forward on this port +if nc -z 127.0.0.1 "$LOCAL_PORT" 2>/dev/null; then + echo "port $LOCAL_PORT already in use — killing stale process..." + fuser -k "$LOCAL_PORT"/tcp 2>/dev/null || true + sleep 1 +fi + +echo "starting kubectl port-forward on 127.0.0.1:$LOCAL_PORT..." +kubectl port-forward -n "$NAMESPACE" "svc/$SERVICE" "$LOCAL_PORT:5432" & +PF_PID=$! + +# Wait for port to be open (up to 15s) +echo "waiting for port $LOCAL_PORT to open..." +for i in $(seq 1 30); do + if nc -z 127.0.0.1 "$LOCAL_PORT" 2>/dev/null; then + echo "port $LOCAL_PORT is open" + break + fi + if ! kill -0 "$PF_PID" 2>/dev/null; then + echo "kubectl port-forward exited unexpectedly" >&2 + exit 1 + fi + sleep 0.5 +done + +if ! nc -z 127.0.0.1 "$LOCAL_PORT" 2>/dev/null; then + echo "port $LOCAL_PORT did not open in time" >&2 + exit 1 +fi + +echo "running integration tests..." +cd "$REPO_ROOT" +cargo test --test integration -- --ignored diff --git a/src/auth.rs b/src/auth.rs index 43a1304..3d38ed7 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -1,19 +1,9 @@ -use crate::error::ProxyError; -use crate::handler::Session; -use crate::pool::ConnectionManager; -use async_trait::async_trait; +//! JWT authentication utilities. + use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; -use futures::SinkExt; -use pgwire::api::auth::{finish_authentication, save_startup_parameters_to_metadata, ServerParameterProvider}; -use pgwire::api::{ClientInfo, PgWireConnectionState}; -use pgwire::error::{PgWireError, PgWireResult}; -use pgwire::messages::startup::Authentication; -use pgwire::messages::{PgWireBackendMessage, PgWireFrontendMessage}; use serde::{Deserialize, Serialize}; -use std::sync::Arc; - -pub const METADATA_USER_ID: &str = "pgwire_supabase_proxy.user_id"; +/// JWT claims extracted from the `sub` field. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct Claims { pub sub: String, @@ -27,6 +17,8 @@ pub struct Claims { pub email: Option, } +/// Validates HS256 JWTs. +#[derive(Clone)] pub struct JwtAuthenticator { jwt_secret: String, } @@ -36,7 +28,8 @@ impl JwtAuthenticator { Self { jwt_secret } } - pub async fn validate_token(&self, token: &str) -> Result { + /// Decode and verify a JWT. Returns the claims on success. + pub async fn validate_token(&self, token: &str) -> Result { let mut validation = Validation::new(Algorithm::HS256); validation.validate_exp = true; @@ -48,116 +41,14 @@ impl JwtAuthenticator { .map_err(|e| { tracing::debug!(error = %e, "JWT validation failed"); match e.kind() { - jsonwebtoken::errors::ErrorKind::ExpiredSignature => ProxyError::JwtExpired, - _ => ProxyError::InvalidJwt(e.to_string()), + jsonwebtoken::errors::ErrorKind::ExpiredSignature => crate::ProxyError::JwtExpired, + _ => crate::ProxyError::InvalidJwt(e.to_string()), } }) .map(|td| td.claims) } } -pub struct StartupHandler { - auth: Arc, - param_provider: Arc, - manager: Arc, - /// Set once in `on_startup` after JWT auth. Shared (via Arc) with `ProxyQueryHandler` - /// so both handlers access the same backend connection. Dropped when the socket closes, - /// which triggers `Session::drop` → connection returned to pool. - session: Arc, -} - -impl StartupHandler { - pub fn new( - auth: Arc, - param_provider: Arc, - manager: Arc, - session: Arc, - ) -> Self { - Self { - auth, - param_provider, - manager, - session, - } - } -} - -impl Clone for StartupHandler { - fn clone(&self) -> Self { - Self { - auth: self.auth.clone(), - param_provider: self.param_provider.clone(), - manager: self.manager.clone(), - session: self.session.clone(), - } - } -} - -#[async_trait] -impl pgwire::api::auth::StartupHandler for StartupHandler -where - S: ServerParameterProvider + 'static, -{ - async fn on_startup( - &self, - client: &mut C, - message: PgWireFrontendMessage, - ) -> PgWireResult<()> - where - C: ClientInfo + futures::Sink + Unpin + Send + Sync, - C::Error: std::fmt::Debug, - PgWireError: From, - { - match message { - PgWireFrontendMessage::Startup(ref startup) => { - save_startup_parameters_to_metadata(client, startup); - client.set_state(PgWireConnectionState::AuthenticationInProgress); - client - .feed(PgWireBackendMessage::Authentication( - Authentication::CleartextPassword, - )) - .await - .map_err(PgWireError::from)?; - client.flush().await.map_err(PgWireError::from)?; - } - PgWireFrontendMessage::PasswordMessageFamily(pwd) => { - let token = pwd.into_password()?.password; - - tracing::info!( - user_prefix = %token.chars().take(20).collect::(), - "connection attempt" - ); - - let claims = self.auth.validate_token(&token).await.map_err(|e| { - tracing::warn!(error = %e, "authentication failed"); - PgWireError::ApiError(Box::new(e)) - })?; - - tracing::info!(user_id = %claims.sub, "authenticated"); - - match self.manager.check_out(&claims.sub).await { - Ok(c) => { - self.session.inner.lock().await.replace(c); - tracing::debug!(user_id = %claims.sub, "backend connection acquired"); - } - Err(e) => { - tracing::error!(error = %e, user_id = %claims.sub, "failed to acquire backend connection"); - return Err(PgWireError::ApiError(Box::new(e))); - } - } - - client - .metadata_mut() - .insert(METADATA_USER_ID.to_string(), claims.sub.clone()); - - finish_authentication(client, self.param_provider.as_ref()).await?; - } - _ => {} - } - Ok(()) - } -} - #[cfg(test)] mod tests { use super::*; @@ -194,19 +85,12 @@ mod tests { #[tokio::test] async fn test_valid_jwt() { let secret = "test-secret-32-chars-minimum!"; - let token = make_test_token(secret, "user-123", 3600); + let token = make_test_token(secret, "550e8400-e29b-41d4-a716-446655440000", 3600); let auth = JwtAuthenticator::new(secret.to_string()); let result = auth.validate_token(&token).await; assert!(result.is_ok()); - assert_eq!(result.unwrap().sub, "user-123"); - } - - #[tokio::test] - async fn test_invalid_jwt() { - let auth = JwtAuthenticator::new("test-secret".to_string()); - let result = auth.validate_token("invalid.token.here").await; - assert!(matches!(result, Err(ProxyError::InvalidJwt(_)))); + assert_eq!(result.unwrap().sub, "550e8400-e29b-41d4-a716-446655440000"); } #[tokio::test] @@ -216,7 +100,14 @@ mod tests { let auth = JwtAuthenticator::new(secret.to_string()); let result = auth.validate_token(&token).await; - assert!(matches!(result, Err(ProxyError::JwtExpired))); + assert!(matches!(result, Err(crate::ProxyError::JwtExpired))); + } + + #[tokio::test] + async fn test_invalid_jwt() { + let auth = JwtAuthenticator::new("test-secret".to_string()); + let result = auth.validate_token("invalid.token.here").await; + assert!(matches!(result, Err(crate::ProxyError::InvalidJwt(_)))); } #[tokio::test] @@ -224,6 +115,6 @@ mod tests { let token = make_test_token("correct-secret-32-chars-minimum", "user-123", 3600); let auth = JwtAuthenticator::new("wrong-secret-32-chars-minimum!!".to_string()); let result = auth.validate_token(&token).await; - assert!(matches!(result, Err(ProxyError::InvalidJwt(_)))); + assert!(matches!(result, Err(crate::ProxyError::InvalidJwt(_)))); } } diff --git a/src/error.rs b/src/error.rs index 4391ce5..55c45b8 100644 --- a/src/error.rs +++ b/src/error.rs @@ -8,27 +8,27 @@ pub enum ProxyError { #[error("JWT expired")] JwtExpired, - #[error("database error: {0}")] - Database(#[from] tokio_postgres::Error), + #[error("protocol violation: {0}")] + ProtocolViolation(String), - #[error("pool error: {0}")] - Pool(#[from] deadpool_postgres::PoolError), + #[error("backend auth error: {0}")] + BackendAuth(String), - #[error("pgwire error: {0}")] - PgWire(#[from] pgwire::error::PgWireError), - - #[error("invalid startup: {0}")] - InvalidStartup(String), + #[error("backend error: {0}")] + BackendError(String), #[error("connection closed")] ConnectionClosed, #[error("encoding error: {0}")] Encoding(String), -} -impl From for pgwire::error::PgWireError { - fn from(e: ProxyError) -> Self { - pgwire::error::PgWireError::ApiError(Box::new(e)) - } + #[error("invalid startup: {0}")] + InvalidStartup(String), + + #[error("TLS error: {0}")] + Tls(String), + + #[error("io error: {0}")] + Io(#[from] std::io::Error), } diff --git a/src/handler.rs b/src/handler.rs deleted file mode 100644 index 69c8a15..0000000 --- a/src/handler.rs +++ /dev/null @@ -1,636 +0,0 @@ -use crate::auth::METADATA_USER_ID; -use crate::error::ProxyError; -use crate::pool::ConnectionManager; -use async_trait::async_trait; -use bytes::Bytes; -use futures::{stream, Sink, Stream}; -use pgwire::api::portal::Portal; -use pgwire::api::query::ExtendedQueryHandler; -use pgwire::api::results::{ - DataRowEncoder, DescribePortalResponse, DescribeStatementResponse, FieldInfo, FieldFormat, - QueryResponse, Response, Tag, -}; -use pgwire::api::stmt::{QueryParser, StoredStatement}; -use pgwire::api::{ClientInfo, Type}; -use pgwire::error::{ErrorInfo, PgWireError, PgWireResult}; -use pgwire::messages::data::DataRow; -use pgwire::messages::PgWireBackendMessage; -use std::fmt::Debug; -use std::pin::Pin; -use std::sync::Arc; -use tokio::sync::Mutex; - -struct ParsedMessages { - columns: Option>>, - data_rows: Vec>, - rows_count: usize, -} - -/// Holds the backend Postgres connection for a client socket, shared between -/// `StartupHandler` (sets it after auth) and `ProxyQueryHandler` (uses it for queries). -/// -/// `Arc` lives for the socket lifetime. When the last `Arc` is dropped -/// (after `process_socket` returns), `Drop` returns the connection to the pool. -/// `RecyclingMethod::Clean` runs `DISCARD ALL` at the next checkout, preventing session state leaks. -pub struct Session { - pub(crate) inner: Arc>>, -} - -impl Session { - pub fn new() -> Self { - Self { - inner: Arc::new(Mutex::new(None)), - } - } -} - -impl Drop for Session { - fn drop(&mut self) { - if let Ok(mut mutex_guard) = self.inner.try_lock() { - let _conn = mutex_guard.take(); - } - } -} - -/// Query handler that shares a single backend connection per socket. -pub struct ProxyQueryHandler { - manager: Arc, - session: Arc, -} - -impl ProxyQueryHandler { - pub fn new(manager: Arc, session: Arc) -> Self { - Self { manager, session } - } - - fn get_user_id(&self, client: &C) -> PgWireResult { - client - .metadata() - .get(METADATA_USER_ID) - .cloned() - .ok_or_else(|| { - PgWireError::ApiError(Box::new(ProxyError::InvalidStartup("no user_id".into()))) - }) - } - - /// Acquire the session connection, run `sql`, restore the connection, return the raw messages. - /// - /// Returns `Err` only for infrastructure failures (no session, pool error). - /// DB-level errors are returned as `Ok(Err(...))` so callers can send them - /// as protocol-level error responses without tearing down the connection. - async fn run_query( - &self, - sql: &str, - fallback_user_id: Option<&str>, - ) -> PgWireResult, tokio_postgres::Error>> { - let backend = { self.session.inner.lock().await.take() }; - - let backend = match (backend, fallback_user_id) { - (Some(c), _) => c, - (None, Some(uid)) => { - tracing::warn!("session has no backend connection, checking out per-query"); - self.manager - .check_out(uid) - .await - .map_err(|e| PgWireError::ApiError(Box::new(e)))? - } - (None, None) => { - return Err(PgWireError::UserError(Box::new(ErrorInfo::new( - "FATAL".into(), - "50000".into(), - "no backend connection in session".into(), - )))); - } - }; - - let messages = backend.simple_query(sql).await; - - { - let mut guard = self.session.inner.lock().await; - if guard.is_none() { - *guard = Some(backend); - } - } - - Ok(messages) - } - - /// Execute `f` with a backend connection borrowed from the session (or checked out - /// per-query), then restore the connection before returning. - async fn with_backend( - &self, - client: &C, - fallback_user_id: Option<&str>, - sql: &str, - ) -> PgWireResult - where - C: ClientInfo, - { - let user_id = self.get_user_id(client)?; - let backend = { self.session.inner.lock().await.take() }; - - let backend = match backend { - Some(c) => c, - None => { - tracing::warn!( - "session has no backend connection, checking out per-query" - ); - self.manager - .check_out(fallback_user_id.unwrap_or(&user_id)) - .await - .map_err(|e| PgWireError::ApiError(Box::new(e)))? - } - }; - - let result = backend.prepare(sql).await; - { - let mut guard = self.session.inner.lock().await; - if guard.is_none() { - *guard = Some(backend); - } - } - - result.map_err(|e| PgWireError::ApiError(Box::new(e))) - } - - /// Parse raw `SimpleQueryMessage`s into columns + encoded rows. - fn parse_messages( - messages: Vec, - ) -> ParsedMessages { - let mut columns: Option>> = None; - let mut data_rows: Vec> = Vec::new(); - let mut rows_count = 0usize; - - for msg in messages { - match msg { - tokio_postgres::SimpleQueryMessage::RowDescription(cols) => { - let fields: Vec = cols - .iter() - .map(|col| { - FieldInfo::new( - col.name().to_string(), - None, - None, - Type::UNKNOWN, - pgwire::api::results::FieldFormat::Text, - ) - }) - .collect(); - columns = Some(Arc::new(fields)); - } - tokio_postgres::SimpleQueryMessage::Row(row) => { - let cols = match &columns { - Some(c) => c.clone(), - None => continue, - }; - let mut encoder = DataRowEncoder::new(cols.clone()); - for col in row.columns() { - let val: Option<&str> = row.get(col.name()); - if let Some(s) = val { - let _ = encoder.encode_field(&s); - } else { - let _ = encoder.encode_field::>(&None); - } - } - data_rows.push(Ok(encoder.take_row())); - rows_count += 1; - } - _ => {} - } - } - - ParsedMessages { columns, data_rows, rows_count } - } - - fn exec_query(messages: Vec) -> Vec { - let ParsedMessages { columns, data_rows, rows_count } = Self::parse_messages(messages); - if let Some(cols) = columns { - let row_stream: Pin> + Send>> = - Box::pin(stream::iter(data_rows)); - let mut qr = QueryResponse::new(cols, row_stream); - qr.set_command_tag(&format!("SELECT {}", rows_count)); - vec![Response::Query(qr)] - } else { - vec![Response::EmptyQuery] - } - } - - fn exec_query_stream(messages: Vec) -> QueryResponse { - let ParsedMessages { columns, data_rows, rows_count } = Self::parse_messages(messages); - match columns { - Some(cols) => { - let row_stream: Pin> + Send>> = - Box::pin(stream::iter(data_rows)); - let mut qr = QueryResponse::new(cols, row_stream); - qr.set_command_tag(&format!("SELECT {}", rows_count)); - qr - } - None => { - let cols = Arc::new(Vec::new()); - let row_stream: Pin> + Send>> = - Box::pin(stream::iter(Vec::new())); - QueryResponse::new(cols, row_stream) - } - } - } - - fn exec_command_tag(messages: Vec) -> Tag { - let mut rows_affected = 0u64; - for msg in messages { - if let tokio_postgres::SimpleQueryMessage::CommandComplete(count) = msg { - rows_affected = count; - } - } - Tag::new("OK").with_rows(rows_affected as usize) - } -} - -impl Clone for ProxyQueryHandler { - fn clone(&self) -> Self { - Self::new(self.manager.clone(), self.session.clone()) - } -} - -#[async_trait] -impl pgwire::api::query::SimpleQueryHandler for ProxyQueryHandler { - async fn do_query(&self, client: &mut C, query: &str) -> PgWireResult> - where - C: ClientInfo + Send + Sync + Unpin, - { - let user_id = self.get_user_id(client)?; - tracing::debug!(user_id = %user_id, query = %query, "do_query"); - - let messages = self.run_query(query, Some(&user_id)).await?; - - match messages { - Ok(msgs) => { - let upper = query.trim().to_uppercase(); - let mut responses = if is_select_query(&upper) { - Self::exec_query(msgs) - } else { - vec![Response::Execution(Self::exec_command_tag(msgs))] - }; - - if upper == "BEGIN" || upper.starts_with("BEGIN ") { - responses.push(Response::TransactionStart(Tag::new("BEGIN"))); - } else if upper == "COMMIT" { - responses.push(Response::TransactionEnd(Tag::new("COMMIT"))); - } else if upper == "ROLLBACK" || upper.starts_with("ABORT") { - responses.push(Response::TransactionEnd(Tag::new("ROLLBACK"))); - } - - Ok(responses) - } - Err(e) => { - tracing::warn!(error = %e, "query error"); - Ok(vec![Response::Error(Box::new(ErrorInfo::new( - "ERROR".into(), - "42000".into(), - e.to_string(), - )))]) - } - } - } -} - -#[async_trait] -impl ExtendedQueryHandler for ProxyQueryHandler { - type Statement = String; - type QueryParser = StringQueryParser; - - fn query_parser(&self) -> Arc { - Arc::new(StringQueryParser) - } - - async fn do_query( - &self, - client: &mut C, - portal: &Portal, - _max_rows: usize, - ) -> PgWireResult - where - C: ClientInfo + Sink + Unpin + Send + Sync, - C::Error: Debug, - PgWireError: From<>::Error>, - { - let _user_id = self.get_user_id(client)?; - let query = portal.statement.statement.clone(); - let q = substitute_params(&query, &portal.parameters); - let upper = query.trim().to_uppercase(); - - let messages = self.run_query(&q, None).await?; - - match messages { - Ok(msgs) => { - if is_select_query(&upper) { - Ok(Response::Query(Self::exec_query_stream(msgs))) - } else { - Ok(Response::Execution(Self::exec_command_tag(msgs))) - } - } - Err(e) => { - tracing::warn!(error = %e, "extended query error"); - Ok(Response::Error(Box::new(ErrorInfo::new( - "ERROR".into(), - "42000".into(), - e.to_string(), - )))) - } - } - } - - /// Override required because `StringQueryParser::get_parameter_types` and - /// `StringQueryParser::get_result_schema` both return empty vecs. - /// The default implementations of `do_describe_statement` / `do_describe_portal` - /// would therefore tell clients there are 0 parameters and 0 result columns, - /// causing "expected 0 parameters but got N" errors and column index panics. - /// By forwarding the describe to the Postgres backend we get the real - /// parameter/column metadata from the prepared statement. - async fn do_describe_statement( - &self, - client: &mut C, - target: &StoredStatement, - ) -> PgWireResult - where - C: ClientInfo + Unpin + Send + Sync, - { - let sql = target.statement.clone(); - let stmt = self.with_backend(client, None, &sql).await?; - - let param_types: Vec = stmt.params().to_vec(); - let fields: Vec = stmt - .columns() - .iter() - .map(|col| { - FieldInfo::new( - col.name().to_string(), - None, - None, - col.type_().clone(), - FieldFormat::Text, - ) - }) - .collect(); - - Ok(DescribeStatementResponse::new(param_types, fields)) - } - - async fn do_describe_portal( - &self, - client: &mut C, - portal: &Portal, - ) -> PgWireResult - where - C: ClientInfo + Unpin + Send + Sync, - { - let sql = portal.statement.statement.clone(); - let stmt = self.with_backend(client, None, &sql).await?; - - let fields: Vec = stmt - .columns() - .iter() - .enumerate() - .map(|(idx, col)| { - let fmt = portal.result_column_format.format_for(idx); - FieldInfo::new(col.name().to_string(), None, None, col.type_().clone(), fmt) - }) - .collect(); - - Ok(DescribePortalResponse::new(fields)) - } -} - -/// QueryParser that returns the SQL string as-is (no actual parsing). -#[derive(Debug, Clone, Default)] -pub struct StringQueryParser; - -#[async_trait] -impl QueryParser for StringQueryParser { - type Statement = String; - - async fn parse_sql( - &self, - _client_info: &C, - query: &str, - _param_types: &[Option], - ) -> PgWireResult - where - C: ClientInfo + Unpin + Send + Sync, - { - Ok(query.to_string()) - } - - fn get_parameter_types(&self, _stmt: &Self::Statement) -> PgWireResult> { - Ok(vec![]) - } - - fn get_result_schema( - &self, - _stmt: &Self::Statement, - _column_format: Option<&pgwire::api::portal::Format>, - ) -> PgWireResult> { - Ok(vec![]) - } -} - -/// Substitute PostgreSQL `$1`, `$2`, ... placeholders with parameter values. -/// Parameters are expected in text format (Bytes encoding a UTF-8 string). -fn substitute_params(sql: &str, params: &[Option]) -> String { - if params.is_empty() { - return sql.to_string(); - } - let mut result = String::with_capacity(sql.len() + params.len() * 16); - let bytes = sql.as_bytes(); - let mut param_idx = 0usize; - let mut i = 0; - - while i < bytes.len() { - if bytes[i] == b'$' { - let start = i; - i += 1; - let mut num = 0usize; - while i < bytes.len() && bytes[i].is_ascii_digit() { - num = num * 10 + (bytes[i] - b'0') as usize; - i += 1; - } - if i > start + 1 && (i >= bytes.len() || !bytes[i].is_ascii_digit()) { - param_idx += 1; - if num == param_idx { - if let Some(Some(p)) = params.get(param_idx - 1) { - if let Ok(s) = std::str::from_utf8(p) { - result.push('\''); - result.push_str(&escape_pg_string(s)); - result.push('\''); - continue; - } - } - result.push_str(&sql[start..i]); - continue; - } else { - i = start + 1; - result.push('$'); - continue; - } - } - i = start + 1; - result.push('$'); - continue; - } - result.push(bytes[i] as char); - i += 1; - } - result -} - -/// Escape a string value for use in a PostgreSQL literal. -pub(crate) fn escape_pg_string(s: &str) -> String { - let mut r = String::with_capacity(s.len() * 2); - for c in s.chars() { - match c { - '\'' => r.push_str("''"), - '\\' => r.push_str("\\\\"), - '\n' => r.push_str("\\n"), - '\r' => r.push_str("\\r"), - '\t' => r.push_str("\\t"), - _ => r.push(c), - } - } - r -} - -fn is_select_query(q: &str) -> bool { - q.starts_with("SELECT") - || q.starts_with("WITH") - || q.starts_with("TABLE") - || q.starts_with("VALUES") -} - -#[cfg(test)] -mod tests { - use super::*; - - // ── escape_pg_string ────────────────────────────────────────────────── - - #[test] - fn test_escape_single_quote() { - assert_eq!(escape_pg_string("it's fine"), "it''s fine"); - } - - #[test] - fn test_escape_backslash() { - assert_eq!(escape_pg_string("C:\\path"), "C:\\\\path"); - } - - #[test] - fn test_escape_newline() { - assert_eq!(escape_pg_string("line1\nline2"), "line1\\nline2"); - } - - #[test] - fn test_escape_cr_tab() { - assert_eq!(escape_pg_string("a\rb\tc"), "a\\rb\\tc"); - } - - #[test] - fn test_escape_empty() { - assert_eq!(escape_pg_string(""), ""); - } - - #[test] - fn test_escape_no_special_chars() { - assert_eq!(escape_pg_string("hello world"), "hello world"); - } - - // ── substitute_params ───────────────────────────────────────────────── - - fn p(s: &str) -> Option { - Some(Bytes::from(s.to_string())) - } - - #[test] - fn test_substitute_basic() { - let sql = "SELECT * FROM t WHERE id = $1"; - let result = substitute_params(sql, &[p("abc")]); - assert_eq!(result, "SELECT * FROM t WHERE id = 'abc'"); - } - - #[test] - fn test_substitute_quote_injection() { - let sql = "SELECT * FROM users WHERE name = $1"; - let result = substitute_params(sql, &[p("O'Brien")]); - assert_eq!(result, "SELECT * FROM users WHERE name = 'O''Brien'"); - } - - #[test] - fn test_substitute_backslash() { - let sql = "SELECT $1"; - let result = substitute_params(sql, &[p("a\\b")]); - assert_eq!(result, "SELECT 'a\\\\b'"); - } - - #[test] - fn test_substitute_multiple_params() { - let sql = "INSERT INTO t (a, b) VALUES ($1, $2)"; - let result = substitute_params(sql, &[p("foo"), p("bar")]); - assert_eq!(result, "INSERT INTO t (a, b) VALUES ('foo', 'bar')"); - } - - #[test] - fn test_substitute_null_param_leaves_placeholder() { - let sql = "SELECT $1"; - let result = substitute_params(sql, &[None]); - assert_eq!(result, "SELECT $1"); - } - - #[test] - fn test_substitute_no_params() { - let sql = "SELECT 1"; - let result = substitute_params(sql, &[]); - assert_eq!(result, "SELECT 1"); - } - - #[test] - fn test_substitute_non_utf8_leaves_placeholder() { - let sql = "SELECT $1"; - let bad_bytes = Some(Bytes::from(vec![0xFF, 0xFE])); - let result = substitute_params(sql, &[bad_bytes]); - assert_eq!(result, "SELECT $1"); - } - - #[test] - fn test_substitute_out_of_order_leaves_unsubstituted() { - // $2 before $1 — both left unsubstituted (sequential-only contract) - let sql = "SELECT $2, $1"; - let result = substitute_params(sql, &[p("first"), p("second")]); - assert_eq!(result, "SELECT $2, $1"); - } - - #[test] - fn test_substitute_repeated_placeholder_second_unsubstituted() { - // $1 twice — only first substitution fires - let sql = "SELECT $1, $1"; - let result = substitute_params(sql, &[p("val")]); - assert_eq!(result, "SELECT 'val', $1"); - } - - #[test] - fn test_substitute_type_cast_delimiter() { - // $1 followed by :: should be substituted - let sql = "SELECT $1::text"; - let result = substitute_params(sql, &[p("hello")]); - assert_eq!(result, "SELECT 'hello'::text"); - } - - // ── is_select_query ─────────────────────────────────────────────────── - - #[test] - fn test_is_select_query_variants() { - assert!(is_select_query("SELECT 1")); - assert!(is_select_query("WITH cte AS (SELECT 1) SELECT * FROM cte")); - assert!(is_select_query("TABLE users")); - assert!(is_select_query("VALUES (1, 2)")); - assert!(!is_select_query("INSERT INTO users VALUES (1)")); - assert!(!is_select_query("UPDATE users SET name = 'x'")); - assert!(!is_select_query("DELETE FROM users")); - } -} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..cc2d563 --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,54 @@ +//! pgwire-supabase-proxy library +//! +//! A byte-forward Postgres proxy that authenticates clients via JWT, +//! opens a backend connection with its own credentials, injects the JWT +//! `sub` claim into the backend session, and then copies bytes transparently. + +mod auth; +mod error; +mod proxy; +mod scram; +mod wire; + +// Re-export Config for use by integration tests and main binary. +pub use auth::{Claims, JwtAuthenticator}; +pub use error::ProxyError; +pub use proxy::serve; + +/// Configuration for the byte-forward proxy. +#[derive(Clone, Debug)] +pub struct Config { + /// Full Postgres connection URL for psp's backend connection. + /// Must include sslmode=require (TLS is mandatory). + pub backend_postgres_url: String, + /// Secret used to validate incoming JWTs (HS256). + pub jwt_secret: String, + /// Address to listen on. + pub listen_addr: String, +} + +impl Config { + pub fn new( + backend_postgres_url: String, + jwt_secret: String, + listen_addr: String, + ) -> Result { + if backend_postgres_url.is_empty() { + return Err(ProxyError::InvalidStartup( + "backend_postgres_url must be non-empty".into(), + )); + } + if jwt_secret.len() < 8 { + return Err(ProxyError::InvalidStartup(format!( + "jwt_secret too short ({} bytes, minimum 8)", + jwt_secret.len() + ))); + } + // No TLS requirement in no-TLS MVP. + Ok(Self { + backend_postgres_url, + jwt_secret, + listen_addr, + }) + } +} diff --git a/src/main.rs b/src/main.rs index 668da24..ec02e4b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,91 +1,25 @@ -mod auth; -mod error; -mod handler; -mod pool; +//! Byte-forward Postgres proxy binary. -use crate::auth::{JwtAuthenticator, StartupHandler}; -use crate::handler::{ProxyQueryHandler, Session}; -use crate::pool::ConnectionManager; -use pgwire::api::auth::DefaultServerParameterProvider; -use pgwire::api::PgWireServerHandlers; +use pgwire_supabase_proxy::{serve, Config}; use std::net::SocketAddr; -use std::sync::Arc; use tokio::net::TcpListener; -use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; - -struct AppFactory { - startup: Arc>, - query: Arc, -} - -impl AppFactory { - fn new(jwt_secret: String, manager: Arc) -> Self { - let auth = Arc::new(JwtAuthenticator::new(jwt_secret)); - let param_provider = DefaultServerParameterProvider::default(); - let session: Arc = Arc::new(Session::new()); - let startup = Arc::new(StartupHandler::new( - auth, - Arc::new(param_provider), - manager.clone(), - session.clone(), - )); - let query = Arc::new(ProxyQueryHandler::new(manager, session)); - - Self { startup, query } - } -} -impl PgWireServerHandlers for AppFactory { - fn startup_handler(&self) -> Arc { - self.startup.clone() - } - - fn simple_query_handler(&self) -> Arc { - self.query.clone() - } - - fn extended_query_handler(&self) -> Arc { - self.query.clone() - } -} #[tokio::main] -async fn main() -> std::result::Result<(), Box> { - tracing_subscriber::registry() - .with( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| "info,pgwire_supabase_proxy=debug".into()), - ) - .with(tracing_subscriber::fmt::layer()) - .init(); - +async fn main() -> std::result::Result<(), Box> { + let backend_postgres_url = + std::env::var("BACKEND_POSTGRES_URL").expect("BACKEND_POSTGRES_URL must be set"); let jwt_secret = std::env::var("SUPABASE_JWT_SECRET").expect("SUPABASE_JWT_SECRET must be set"); - let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set"); let listen_addr: SocketAddr = std::env::var("LISTEN_ADDR") .unwrap_or_else(|_| "0.0.0.0:5432".to_string()) .parse() .expect("invalid LISTEN_ADDR"); - let pool_size: usize = std::env::var("POOL_SIZE") - .unwrap_or_else(|_| "10".to_string()) - .parse() - .expect("invalid POOL_SIZE"); let listener = TcpListener::bind(listen_addr).await?; - let manager = Arc::new(ConnectionManager::new(database_url, pool_size)); - tracing::info!(addr = %listen_addr, "starting pgwire-supabase-proxy"); - - loop { - let (socket, addr) = listener.accept().await?; - tracing::info!(addr = %addr, "connection accepted"); + let config = Config::new(backend_postgres_url, jwt_secret, listen_addr.to_string())?; - let factory = Arc::new(AppFactory::new(jwt_secret.clone(), manager.clone())); - tokio::spawn(async move { - let result = pgwire::tokio::process_socket(socket, None, factory.clone()).await; - if let Err(e) = result { - tracing::error!(error = %e, "connection error"); - } - // Arc is dropped here → Arc refcount hits 0 - // → Session::drop runs → DISCARD ALL on the backend connection. - }); - } + serve(config, listener, async { + let _ = tokio::signal::ctrl_c().await; + }) + .await } diff --git a/src/pool.rs b/src/pool.rs deleted file mode 100644 index 59283c1..0000000 --- a/src/pool.rs +++ /dev/null @@ -1,94 +0,0 @@ -use crate::error::ProxyError; -use crate::handler::escape_pg_string; -use deadpool_postgres::{Config, ManagerConfig, Pool, RecyclingMethod}; -use tokio::sync::Mutex; - -/// Manages backend Postgres connection pools per user. -pub struct ConnectionManager { - pools: std::sync::Arc>>, - db_url: String, - max_connections: usize, -} - -impl ConnectionManager { - pub fn new(database_url: String, max_connections: usize) -> Self { - Self { - pools: std::sync::Arc::new(Mutex::new(lru::LruCache::new( - std::num::NonZeroUsize::new(1024).unwrap(), - ))), - db_url: database_url, - max_connections, - } - } - - /// Get or create a pool for the given user_id. - pub async fn get_pool(&self, user_id: &str) -> Result { - let mut pools = self.pools.lock().await; - if let Some(pool) = pools.get(user_id) { - return Ok(pool.clone()); - } - - let mut cfg = Config::new(); - cfg.url = Some(self.db_url.clone()); - cfg.manager = Some(ManagerConfig { - recycling_method: RecyclingMethod::Clean, - }); - cfg.pool = Some(deadpool_postgres::PoolConfig::new(self.max_connections)); - let pool = cfg - .create_pool( - Some(deadpool_postgres::Runtime::Tokio1), - tokio_postgres::NoTls, - ) - .map_err(|e| ProxyError::InvalidStartup(format!("failed to create pool: {}", e)))?; - - pools.push(user_id.to_string(), pool.clone()); - Ok(pool) - } - - /// Check out a connection and set RLS context. - pub async fn check_out(&self, user_id: &str) -> Result { - let pool = self.get_pool(user_id).await?; - let client = pool.get().await?; - - // Set role to authenticated (bypassrls=false → RLS applies) - client.simple_query("SET ROLE authenticated").await?; - - // Set request.jwt.claim.sub so auth.uid() works - client - .simple_query(&format!( - "SET request.jwt.claim.sub = '{}'", - escape_user_id(user_id) - )) - .await?; - - tracing::debug!(user_id = %user_id, "RLS context set"); - Ok(client) - } -} - -/// Escape a user_id for safe interpolation into a SET statement literal. -/// Delegates to `escape_pg_string` to keep a single escaping SSOT. -pub(crate) fn escape_user_id(user_id: &str) -> String { - escape_pg_string(user_id) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_escape_user_id_normal_uuid() { - let uid = "550e8400-e29b-41d4-a716-446655440000"; - assert_eq!(escape_user_id(uid), uid); - } - - #[test] - fn test_escape_user_id_single_quote() { - assert_eq!(escape_user_id("user'123"), "user''123"); - } - - #[test] - fn test_escape_user_id_multiple_quotes() { - assert_eq!(escape_user_id("a'b'c"), "a''b''c"); - } -} diff --git a/src/proxy.rs b/src/proxy.rs new file mode 100644 index 0000000..456f239 --- /dev/null +++ b/src/proxy.rs @@ -0,0 +1,342 @@ +//! Byte-forward proxy core. + +use crate::auth::{Claims, JwtAuthenticator}; +use crate::error::ProxyError; +use crate::scram; +use crate::wire; +use std::collections::HashMap; +use std::net::SocketAddr; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + + + +/// Start the byte-forward proxy server. +pub async fn serve( + config: crate::Config, + listener: TcpListener, + shutdown: impl std::future::Future + Send + 'static, +) -> std::result::Result<(), Box> { + let _ = tracing_subscriber::registry() + .with( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info,pgwire_supabase_proxy=debug".into()), + ) + .with(tracing_subscriber::fmt::layer()) + .try_init(); + + let addr = listener.local_addr()?; + tracing::info!(addr = %addr, "starting pgwire-supabase-proxy"); + + tokio::pin!(shutdown); + + loop { + tokio::select! { + result = listener.accept() => { + let (socket, peer_addr) = result?; + tracing::info!(peer_addr = %peer_addr, "connection accepted"); + + let config = config.clone(); + tokio::spawn(async move { + let result = handle_connection(socket, peer_addr, &config).await; + if let Err(e) = result { + tracing::error!(error = %e, peer_addr = %peer_addr, "connection error"); + } + }); + } + _ = &mut shutdown => { + tracing::info!("shutdown signal received, stopping"); + break; + } + } + } + + Ok(()) +} + +/// Handle one client connection. +async fn handle_connection( + mut client: tokio::net::TcpStream, + peer_addr: SocketAddr, + config: &crate::Config, +) -> std::result::Result<(), Box> { + let start = std::time::Instant::now(); + + // Step 2: Handle SSLRequest + let mut first_msg_len_buf = [0u8; 4]; + client.read_exact(&mut first_msg_len_buf).await?; + let first_msg_len = u32::from_be_bytes(first_msg_len_buf); + + let msg_len = if first_msg_len == 8 { + let mut code_buf = [0u8; 4]; + client.read_exact(&mut code_buf).await?; + let code = u32::from_be_bytes(code_buf); + if code == 80877103 { + client.write_all(b"N").await?; + tracing::debug!(peer_addr = %peer_addr, "SSL request rejected"); + let mut len_buf = [0u8; 4]; + client.read_exact(&mut len_buf).await?; + u32::from_be_bytes(len_buf) + } else { + first_msg_len + } + } else { + first_msg_len + }; + + // Step 3: Parse StartupMessage + let mut startup_buf = vec![0u8; (msg_len - 4) as usize]; + client.read_exact(&mut startup_buf).await?; + let startup = wire::parse_startup_body(msg_len, &startup_buf)?; + let user = startup + .params + .get("user") + .cloned() + .ok_or_else(|| ProxyError::ProtocolViolation("StartupMessage missing user".into()))?; + let database = startup + .params + .get("database") + .cloned() + .unwrap_or_else(|| "postgres".to_string()); + + tracing::debug!(peer_addr = %peer_addr, user = %user, database = %database, "startup received"); + + // Step 4: Authenticate client + wire::write_authentication_cleartext_password(&mut client).await?; + let password = wire::read_password_message(&mut client).await?; + + // Step 5: Verify JWT + let auth = JwtAuthenticator::new(config.jwt_secret.clone()); + let claims: Claims = match auth.validate_token(&password).await { + Ok(c) => c, + Err(e) => { + tracing::warn!(peer_addr = %peer_addr, error = %e, "JWT verification failed"); + wire::write_error_response(&mut client, "28P01", "JWT verification failed").await?; + return Ok(()); + } + }; + let jwt_sub = claims.sub.clone(); + tracing::info!(peer_addr = %peer_addr, user_id = %jwt_sub, "client authenticated"); + + // Validate sub claim + if jwt_sub.len() > 128 || jwt_sub.bytes().any(|b| b == 0) { + wire::write_error_response(&mut client, "28P01", "invalid sub claim").await?; + return Ok(()); + } + if !jwt_sub + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') + { + wire::write_error_response(&mut client, "28P01", "invalid sub claim format").await?; + return Ok(()); + } + + // Step 6: Open backend TCP + let (backend_host, backend_port, backend_user, backend_password, backend_db) = + parse_backend_url(&config.backend_postgres_url)?; + + let mut backend = match tokio::net::TcpStream::connect((backend_host.as_str(), backend_port)).await { + Ok(s) => s, + Err(e) => { + tracing::error!(peer_addr = %peer_addr, error = %e, "backend TCP connect failed"); + wire::write_error_response(&mut client, "08001", &format!("backend connection failed: {}", e)).await?; + return Ok(()); + } + }; + tracing::debug!(peer_addr = %peer_addr, backend = %backend_host, "backend TCP opened"); + + // Step 7: Send backend StartupMessage (plain TCP, no TLS) + let mut backend_params = HashMap::new(); + backend_params.insert("user".into(), backend_user.clone()); + backend_params.insert("database".into(), backend_db.clone()); + backend_params.insert("application_name".into(), format!("psp/{}", user)); + backend_params.insert("client_encoding".into(), "UTF8".into()); + wire::write_startup_message(&mut backend, &backend_params).await?; + + // Step 9: Handle backend auth + let auth_method = wire::read_authentication_method(&mut backend).await?; + match auth_method { + wire::AuthMethod::Ok => { + tracing::debug!(peer_addr = %peer_addr, "backend auth: OK"); + } + wire::AuthMethod::CleartextPassword => { + wire::write_password_message(&mut backend, &backend_password).await?; + } + wire::AuthMethod::Sasl { mechanisms } => { + if !mechanisms.contains(&"SCRAM-SHA-256".to_string()) { + return Err(Box::new(ProxyError::BackendAuth(format!( + "unsupported SASL mechanisms: {:?}", + mechanisms + )))); + } + scram::scram_sha_256_authenticate(&mut backend, &backend_user, &backend_password) + .await + .map_err(|e| Box::new(e) as Box)?; + } + wire::AuthMethod::Md5Password { .. } => { + return Err(Box::new(ProxyError::BackendAuth( + "MD5 auth not supported".into(), + ))); + } + } + + // Step 10: Drain backend until ReadyForQuery + let mut backend_params_response = Vec::new(); + loop { + let msg = wire::read_backend_message(&mut backend).await?; + match msg { + wire::BackendMessage::ReadyForQuery { .. } => break, + wire::BackendMessage::ParameterStatus { key, value } => { + backend_params_response.push((key, value)); + } + wire::BackendMessage::BackendKeyData { process_id, .. } => { + tracing::debug!( + peer_addr = %peer_addr, + backend_pid = process_id, + "backend key data received" + ); + } + wire::BackendMessage::ErrorResponse { + severity, + code, + message, + } => { + return Err(Box::new(ProxyError::BackendError(format!( + "{} {}: {}", + severity.unwrap_or_default(), + code.unwrap_or_default(), + message + )))); + } + wire::BackendMessage::Unknown { .. } => {} + } + } + + // Step 11: Inject JWT claim + let escaped_sub = escape_pg_string(&jwt_sub); + let set_config_sql = format!( + "SELECT set_config('request.jwt.claim.sub', E'{}', false); SET ROLE authenticated;", + escaped_sub + ); + wire::write_query(&mut backend, &set_config_sql).await?; + + loop { + let msg = wire::read_backend_message(&mut backend).await?; + match msg { + wire::BackendMessage::ReadyForQuery { .. } => break, + wire::BackendMessage::ErrorResponse { code, message, .. } => { + tracing::error!( + peer_addr = %peer_addr, + code = ?code, + message = %message, + "set_config/ROLE failed" + ); + wire::write_error_response( + &mut client, + &code.unwrap_or_else(|| "08006".into()), + &format!("backend session setup failed: {}", message), + ) + .await?; + return Ok(()); + } + wire::BackendMessage::Unknown { .. } => {} + _ => {} + } + } + + // Step 12: Complete client startup + wire::write_authentication_ok(&mut client).await?; + for (key, value) in &backend_params_response { + wire::write_parameter_status(&mut client, key, value).await?; + } + let client_pid: i32 = rand::random(); + let client_secret: i32 = rand::random(); + wire::write_backend_key_data(&mut client, client_pid, client_secret).await?; + wire::write_ready_for_query(&mut client, b'I').await?; + client.flush().await?; + + tracing::info!( + peer_addr = %peer_addr, + user_id = %jwt_sub, + "session ready — entering byte-forward mode" + ); + + // Step 13: Byte-forward (bidirectional, concurrent) + let result = tokio::io::copy_bidirectional(&mut client, &mut backend).await; + + match result { + Ok((bytes_to_backend, bytes_to_client)) => { + tracing::info!( + peer_addr = %peer_addr, + user_id = %jwt_sub, + duration_ms = start.elapsed().as_millis() as u64, + bytes_to_backend, + bytes_to_client, + close_reason = "both sides closed", + "connection closed" + ); + } + Err(e) => { + tracing::error!( + peer_addr = %peer_addr, + user_id = %jwt_sub, + duration_ms = start.elapsed().as_millis() as u64, + error = %e, + "connection closed with error" + ); + } + } + + Ok(()) +} + +/// Parse a backend Postgres URL into its components. +/// Accepts both `postgresql://` and `postgres://` schemes. +fn parse_backend_url(url: &str) -> Result<(String, u16, String, String, String), ProxyError> { + let url = url + .trim_start_matches("postgresql://") + .trim_start_matches("postgres://"); + let (creds, rest) = url + .split_once('@') + .ok_or_else(|| ProxyError::InvalidStartup("backend URL missing '@'".into()))?; + let (user, password) = creds + .split_once(':') + .ok_or_else(|| ProxyError::InvalidStartup("backend URL missing password".into()))?; + let (host_port, db_and_query) = rest + .split_once('/') + .ok_or_else(|| ProxyError::InvalidStartup("backend URL missing '/db'".into()))?; + let (host_port, _query) = host_port.split_once('?').unwrap_or((host_port, "")); + let (host, port_str) = host_port.split_once(':').unwrap_or((host_port, "5432")); + let port: u16 = port_str + .parse() + .map_err(|_| ProxyError::InvalidStartup("invalid backend port".into()))?; + // Database name is the path component before any '?' + let database = db_and_query + .split('?') + .next() + .filter(|s| !s.is_empty()) + .unwrap_or("postgres"); + Ok(( + host.to_string(), + port, + user.to_string(), + password.to_string(), + database.to_string(), + )) +} + +fn escape_pg_string(s: &str) -> String { + let mut r = String::with_capacity(s.len() * 2); + for c in s.chars() { + match c { + '\'' => r.push_str("''"), + '\\' => r.push_str("\\\\"), + '\n' => r.push_str("\\n"), + '\r' => r.push_str("\\r"), + '\t' => r.push_str("\\t"), + _ => r.push(c), + } + } + r +} diff --git a/src/scram.rs b/src/scram.rs new file mode 100644 index 0000000..814398d --- /dev/null +++ b/src/scram.rs @@ -0,0 +1,437 @@ +//! SCRAM-SHA-256 client authentication helper. + +use crate::error::ProxyError; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use hmac::{Hmac, Mac}; +use rand::Rng; +use sha2::Sha256; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +type HmacSha256 = Hmac; + +const SHA256_NAME: &str = "SCRAM-SHA-256"; + +/// Perform SCRAM-SHA-256 authentication with the backend. +pub async fn scram_sha_256_authenticate( + stream: &mut S, + username: &str, + password: &str, +) -> Result<(), ProxyError> +where + S: AsyncReadExt + AsyncWriteExt + Unpin, +{ + // Generate nonce before first .await (ThreadRng is !Send) + let client_nonce: String = { + let mut rng = rand::thread_rng(); + (0..18) + .map(|_| { + let b: u8 = rng.gen(); + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789" + .chars() + .nth((b % 62) as usize) + .unwrap() + }) + .collect() + }; // rng dropped here, before first .await + + // Step 1: ClientFirst — GS2 header is "n,," (no channel binding) + let client_first_bare = format!("n={},r={}", username, client_nonce); + let client_first = format!("n,,{}", client_first_bare); + send_sasl_initial_response(stream, SHA256_NAME, client_first.as_bytes()).await?; + + // Step 2: ServerFirst + let server_first_raw = read_sasl_continue(stream).await?; + let server_first_str = std::str::from_utf8(&server_first_raw) + .map_err(|_| ProxyError::BackendAuth("invalid UTF-8 in server-first".into()))?; + + let sf = parse_server_first(server_first_str)?; + + if !sf.server_nonce.starts_with(&client_nonce) || sf.server_nonce.len() <= client_nonce.len() { + return Err(ProxyError::BackendAuth( + "server nonce doesn't start with client nonce".into(), + )); + } + + // Step 3: ClientFinal + let client_final_without_proof = format!("c=biws,r={}", sf.server_nonce); + + let client_proof = compute_client_proof( + password, + client_first_bare.as_bytes(), + server_first_str.as_bytes(), + client_final_without_proof.as_bytes(), + &sf.salt, + sf.iteration_count, + )?; + + let client_final_message = format!(",{}", client_proof); + let full_client_final = format!("{}{}", client_final_without_proof, client_final_message); + + send_sasl_response(stream, full_client_final.as_bytes()).await?; + + // Step 4: ServerSignature — server sends AuthenticationSASLFinal (type 12) + let server_final_raw = read_sasl_final(stream).await?; + let server_final_str = std::str::from_utf8(&server_final_raw) + .map_err(|_| ProxyError::BackendAuth("invalid UTF-8 in server-final".into()))?; + + if let Some(server_sig) = server_final_str.strip_prefix("v=") { + let expected_sig = compute_server_signature( + password, + client_first_bare.as_bytes(), + server_first_str.as_bytes(), + client_final_without_proof.as_bytes(), + &sf.salt, + sf.iteration_count, + )?; + + if server_sig != expected_sig { + return Err(ProxyError::BackendAuth("server signature mismatch".into())); + } + } else if let Some(err_msg) = server_final_str.strip_prefix("e=") { + return Err(ProxyError::BackendAuth(format!( + "server error: {}", + err_msg + ))); + } else { + return Err(ProxyError::BackendAuth(format!( + "unexpected server-final: {}", + server_final_str + ))); + } + + tracing::debug!(username = %username, "SCRAM authentication successful"); + Ok(()) +} + +// ─── SCRAM internals ───────────────────────────────────────────────────────── + +struct ServerFirst { + salt: Vec, + iteration_count: u32, + server_nonce: String, +} + +fn parse_server_first(s: &str) -> Result { + let mut salt = None; + let mut iter_count = None; + let mut server_nonce = None; + + // The nonce in r= may contain commas, so split from the right: after the final ,s= or ,i= + if let Some(r_pos) = s.find("r=") { + // Extract r= value: from "r=" up to the last ",s=" or ",i=" + let after_r = &s[r_pos + 2..]; + let end = after_r + .rfind(",s=") + .or(after_r.rfind(",i=")) + .unwrap_or(after_r.len()); + server_nonce = Some(after_r[..end].to_string()); + } + + // Remaining attrs: s= and i= (after the nonce) + if let Some(rest) = s.split(",s=").nth(1) { + let parts: Vec<&str> = rest.splitn(2, ",i=").collect(); + salt = Some(base64_decode(parts[0]).map_err(|e| ProxyError::BackendAuth(e.to_string()))?); + if parts.len() > 1 { + iter_count = Some( + parts[1] + .parse() + .map_err(|_| ProxyError::BackendAuth("invalid iteration count".into()))?, + ); + } + } + // Handle i= without s= (fallback) + if iter_count.is_none() { + if let Some(rest) = s.split(",i=").nth(1) { + iter_count = Some( + rest.parse() + .map_err(|_| ProxyError::BackendAuth("invalid iteration count".into()))?, + ); + } + } + + Ok(ServerFirst { + salt: salt.ok_or_else(|| ProxyError::BackendAuth("missing salt".into()))?, + iteration_count: iter_count + .ok_or_else(|| ProxyError::BackendAuth("missing iteration count".into()))?, + server_nonce: server_nonce + .ok_or_else(|| ProxyError::BackendAuth("missing server nonce".into()))?, + }) +} + +fn compute_client_proof( + password: &str, + client_first_bare: &[u8], + server_first: &[u8], + client_final_message_without_proof: &[u8], + salt: &[u8], + iteration_count: u32, +) -> Result { + let normalized_password = normalize_password(password); + let salted_password = hi(&normalized_password, salt, iteration_count)?; + + let client_key = hmac_sign(&salted_password, b"Client Key"); + let stored_key = sha256_hash(&client_key); + + let auth_message: Vec = join_bytes(&[ + client_first_bare, + server_first, + client_final_message_without_proof, + ]); + + let client_signature = hmac_sign(&stored_key, &auth_message); + + let mut client_proof = vec![0u8; client_key.len()]; + for i in 0..client_key.len() { + client_proof[i] = client_key[i] ^ client_signature[i]; + } + + Ok(base64_encode(&client_proof)) +} + +fn compute_server_signature( + password: &str, + client_first_bare: &[u8], + server_first: &[u8], + client_final_message_without_proof: &[u8], + salt: &[u8], + iteration_count: u32, +) -> Result { + let normalized_password = normalize_password(password); + let salted_password = hi(&normalized_password, salt, iteration_count)?; + + let server_key = hmac_sign(&salted_password, b"Server Key"); + + let auth_message: Vec = join_bytes(&[ + client_first_bare, + server_first, + client_final_message_without_proof, + ]); + + let server_signature = hmac_sign(&server_key, &auth_message); + + Ok(base64_encode(&server_signature)) +} + +/// PBKDF2-HMAC-SHA256 key derivation. +fn hi(password: &[u8], salt: &[u8], iterations: u32) -> Result, ProxyError> { + let mut result = vec![0u8; 32]; + let mut u = vec![0u8; 32]; + + let mut mac = + HmacSha256::new_from_slice(password).map_err(|e| ProxyError::BackendAuth(e.to_string()))?; + mac.update(salt); + mac.update(&1u32.to_be_bytes()); + u.copy_from_slice(&mac.finalize().into_bytes()); + result.copy_from_slice(&u); // XOR in U1 (result is zeroed, so copy = XOR) + + for _ in 2..=iterations { + let mut mac = HmacSha256::new_from_slice(password) + .map_err(|e| ProxyError::BackendAuth(e.to_string()))?; + mac.update(&u); + u.copy_from_slice(&mac.finalize().into_bytes()); + for i in 0..32 { + result[i] ^= u[i]; + } + } + + Ok(result) +} + +fn hmac_sign(key: &[u8], data: &[u8]) -> Vec { + let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts any key size"); + mac.update(data); + mac.finalize().into_bytes().to_vec() +} + +fn sha256_hash(data: &[u8]) -> Vec { + use sha2::Digest; + Sha256::new().chain_update(data).finalize().to_vec() +} + +fn normalize_password(password: &str) -> Vec { + // RFC 5802: normalize according to SASLprep profile + // For simplicity, we use the password as-is (Postgres SCRAM accepts this) + password.as_bytes().to_vec() +} + +/// Join byte slices with a separator. +fn join_bytes(parts: &[&[u8]]) -> Vec { + let sep = b','; + let total: usize = parts.iter().map(|p| p.len()).sum::() + parts.len().saturating_sub(1); + let mut result = Vec::with_capacity(total); + for (i, part) in parts.iter().enumerate() { + if i > 0 { + result.push(sep); + } + result.extend_from_slice(part); + } + result +} + +fn base64_encode(data: &[u8]) -> String { + BASE64.encode(data) +} + +fn base64_decode(s: &str) -> Result, &'static str> { + BASE64 + .decode(s) + .map_err(|_| "invalid base64") +} + +// ─── Wire framing helpers ──────────────────────────────────────────────────── + +async fn send_sasl_initial_response( + stream: &mut S, + mechanism: &str, + initial_response: &[u8], +) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'p'); + // length = 4 (self) + mechanism + null + Int32(initial_response_len) + initial_response + let response_len = 4 + mechanism.len() + 1 + 4 + initial_response.len(); + buf.extend_from_slice(&(response_len as u32).to_be_bytes()); + buf.extend_from_slice(mechanism.as_bytes()); + buf.push(0); + if initial_response.is_empty() { + buf.extend_from_slice(&(-1i32).to_be_bytes()); + } else { + buf.extend_from_slice(&(initial_response.len() as i32).to_be_bytes()); + buf.extend_from_slice(initial_response); + } + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +async fn send_sasl_response(stream: &mut S, data: &[u8]) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'p'); + let len: u32 = 4 + data.len() as u32; + buf.extend_from_slice(&len.to_be_bytes()); + buf.extend_from_slice(data); + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +async fn read_sasl_auth_message( + stream: &mut S, + expected_type: u32, +) -> Result, ProxyError> +where + S: AsyncReadExt + Unpin, +{ + let mut type_buf = [0u8; 1]; + stream.read_exact(&mut type_buf).await?; + if type_buf[0] != b'R' { + return Err(ProxyError::ProtocolViolation(format!( + "expected SASL auth message (R), got {:02x}", + type_buf[0] + ))); + } + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).await?; + let len = u32::from_be_bytes(len_buf); + let mut body = vec![0u8; (len - 4) as usize]; + stream.read_exact(&mut body).await?; + let auth_type = u32::from_be_bytes([body[0], body[1], body[2], body[3]]); + if auth_type != expected_type { + return Err(ProxyError::ProtocolViolation(format!( + "expected SASL auth type {}, got {}", + expected_type, auth_type + ))); + } + Ok(body[4..].to_vec()) +} + +/// Read AuthenticationSASLContinue (type 11) from the backend. +async fn read_sasl_continue(stream: &mut S) -> Result, ProxyError> +where + S: AsyncReadExt + Unpin, +{ + read_sasl_auth_message(stream, 11).await +} + +/// Read AuthenticationSASLFinal (type 12) from the backend. +async fn read_sasl_final(stream: &mut S) -> Result, ProxyError> +where + S: AsyncReadExt + Unpin, +{ + read_sasl_auth_message(stream, 12).await +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_base64_roundtrip() { + let cases: &[&[u8]] = &[b"a", b"ab", b"abc", b"Hello, World!", b"\x00\xff\xfe\xfd"]; + for case in cases { + let encoded = base64_encode(case); + let decoded = base64_decode(&encoded).unwrap(); + assert_eq!(decoded.as_slice(), *case, "roundtrip failed for {:?}", case); + } + // Also test empty explicitly + let encoded = base64_encode(b""); + assert_eq!(base64_decode(&encoded).unwrap(), b""); + } + + #[test] + fn test_parse_server_first_valid() { + let s = + "r=fyko+d2lbbFgONe9WqKkE2qtVdgo,+5qdLY9Rw=,s=QSXCRQD6Yt6AS+kWSMEpqhGkg5e/klE+,i=4096"; + let sf = parse_server_first(s).unwrap(); + assert_eq!(sf.server_nonce, "fyko+d2lbbFgONe9WqKkE2qtVdgo,+5qdLY9Rw="); + assert_eq!(sf.iteration_count, 4096); + } + + #[test] + fn test_parse_server_first_missing_fields() { + assert!(parse_server_first("r=nonce").is_err()); + assert!(parse_server_first("s=salt").is_err()); + assert!(parse_server_first("i=4096").is_err()); + } + + #[test] + fn test_sha256_hash_known() { + use sha2::Digest; + // Known SHA256 of "test" + let result = sha256_hash(b"test"); + let expected = Sha256::digest(b"test"); + assert_eq!(result, expected.to_vec()); + } + + #[test] + fn test_hmac_sign_deterministic() { + let sig1 = hmac_sign(b"key", b"data"); + let sig2 = hmac_sign(b"key", b"data"); + assert_eq!(sig1, sig2); + assert_eq!(sig1.len(), 32); // SHA256 output = 32 bytes + } + + #[test] + fn test_hi_includes_first_iteration() { + // Regression test: hi() must XOR in U1 (the first HMAC iteration). + // Without this, only iterations 2..n are XORed, giving a wrong result. + // Test vector from RFC 6070 / test vectors for PBKDF2-SHA256: + // password="password", salt="salt", c=4096, DK=120fb06c... + let password = b"password"; + let salt = b"salt"; + let iterations = 4096; + let result = hi(password, salt, iterations).expect("hi should succeed"); + // RFC 6070 test vector: PBKDF2-SHA256("password", "salt", 4096) + assert_eq!( + &result[..4], + &[0xc5, 0xe4, 0x78, 0xd5], + "hi() must XOR in U1 (first iteration); without it the result is wrong" + ); + } + +} diff --git a/src/wire.rs b/src/wire.rs new file mode 100644 index 0000000..82e72d3 --- /dev/null +++ b/src/wire.rs @@ -0,0 +1,439 @@ +//! Minimal Postgres wire protocol message codec. +//! +//! This module implements the subset of the Postgres wire protocol needed for +//! the byte-forward proxy: client auth (JWT), backend auth (SCRAM), and +//! handshake message exchange. After handshake, all bytes are forwarded transparently. + +use crate::error::ProxyError; +use std::collections::HashMap; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +// ─── Client → Proxy messages ───────────────────────────────────────────────── + +/// Represents a parsed StartupMessage. +#[derive(Debug)] +#[allow(dead_code)] +pub struct StartupMessage { + pub protocol_version: u32, + pub params: HashMap, +} + +/// Parse the body of a StartupMessage (length already read). +pub fn parse_startup_body(_msg_len: u32, buf: &[u8]) -> Result { + let protocol_version = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]); + + // Parse null-terminated key=value pairs + let mut params = HashMap::new(); + let mut i = 4; + while i + 1 < buf.len() { + if buf[i] == 0 && buf[i + 1] == 0 { + break; // Final null terminator + } + // Read key + let key_start = i; + while i < buf.len() && buf[i] != 0 { + i += 1; + } + let key = std::str::from_utf8(&buf[key_start..i]) + .map_err(|_| ProxyError::ProtocolViolation("invalid UTF-8 in startup key".into()))? + .to_string(); + i += 1; + if i >= buf.len() { + break; + } + + // Read value + let value_start = i; + while i < buf.len() && buf[i] != 0 { + i += 1; + } + let value = std::str::from_utf8(&buf[value_start..i]) + .map_err(|_| ProxyError::ProtocolViolation("invalid UTF-8 in startup value".into()))? + .to_string(); + i += 1; + + params.insert(key, value); + } + + Ok(StartupMessage { + protocol_version, + params, + }) +} + +/// Read a PasswordMessage from the client. +/// Wire format: Byte1('p') + Int32(len) + String(password, null-terminated) +pub async fn read_password_message(stream: &mut S) -> Result +where + S: AsyncReadExt + Unpin, +{ + let mut type_buf = [0u8; 1]; + stream.read_exact(&mut type_buf).await?; + if type_buf[0] != b'p' { + return Err(ProxyError::ProtocolViolation(format!( + "expected PasswordMessage ('p'), got 0x{:02x}", + type_buf[0] + ))); + } + let len = read_message_length(stream).await?; + let mut buf = vec![0u8; (len - 4) as usize]; + stream.read_exact(&mut buf).await?; + // Strip null terminator + let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len()); + let password = std::str::from_utf8(&buf[..end]) + .map_err(|_| ProxyError::ProtocolViolation("invalid UTF-8 in password".into()))? + .to_string(); + Ok(password) +} + +// ─── Proxy → Client messages ──────────────────────────────────────────────── + +/// Write AuthenticationCleartextPassword (R, type=3). +pub async fn write_authentication_cleartext_password(stream: &mut S) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = [0u8; 9]; // 'R' + len(4) + auth_type(4) + buf[0] = b'R'; + buf[1..5].copy_from_slice(&8u32.to_be_bytes()); // length = 4 (self) + 4 (auth_type) = 8 + buf[5..9].copy_from_slice(&3u32.to_be_bytes()); + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +/// Write AuthenticationOk (R, type=0). +pub async fn write_authentication_ok(stream: &mut S) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = [0u8; 9]; + buf[0] = b'R'; + buf[1..5].copy_from_slice(&8u32.to_be_bytes()); // length = 4 (self) + 4 (auth_type) = 8 + buf[5..9].copy_from_slice(&0u32.to_be_bytes()); + stream.write_all(&buf).await?; + Ok(()) +} + +/// Write a ParameterStatus (S) message. +pub async fn write_parameter_status( + stream: &mut S, + key: &str, + value: &str, +) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'S'); + let len_pos = buf.len(); + buf.extend_from_slice(&[0, 0, 0, 0]); + buf.extend_from_slice(key.as_bytes()); + buf.push(0); + buf.extend_from_slice(value.as_bytes()); + buf.push(0); + // length = buf.len() - 1 (excludes the type byte 'S') + let len = (buf.len() - 1) as u32; + buf[len_pos..len_pos + 4].copy_from_slice(&len.to_be_bytes()); + stream.write_all(&buf).await?; + Ok(()) +} + +/// Write BackendKeyData (K). +pub async fn write_backend_key_data( + stream: &mut S, + process_id: i32, + secret_key: i32, +) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = [0u8; 13]; + buf[0] = b'K'; + buf[1..5].copy_from_slice(&12u32.to_be_bytes()); + buf[5..9].copy_from_slice(&(process_id as u32).to_be_bytes()); + buf[9..13].copy_from_slice(&(secret_key as u32).to_be_bytes()); + stream.write_all(&buf).await?; + Ok(()) +} + +/// Write ReadyForQuery (Z) with transaction status byte. +pub async fn write_ready_for_query(stream: &mut S, status: u8) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = [0u8; 6]; + buf[0] = b'Z'; + buf[1..5].copy_from_slice(&5u32.to_be_bytes()); + buf[5] = status; + stream.write_all(&buf).await?; + Ok(()) +} + +/// Write an ErrorResponse (E). +pub async fn write_error_response( + stream: &mut S, + sqlstate: &str, + message: &str, +) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'E'); + let len_pos = buf.len(); + buf.extend_from_slice(&[0, 0, 0, 0]); + + // Field: Severity + buf.push(b'S'); + buf.extend_from_slice(b"FATAL"); + buf.push(0); + // Field: SQLSTATE + buf.push(b'C'); + buf.extend_from_slice(sqlstate.as_bytes()); + buf.push(0); + // Field: Message + buf.push(b'M'); + buf.extend_from_slice(message.as_bytes()); + buf.push(0); + // Terminator + buf.push(0); + + // length = buf.len() - 1 (excludes the type byte 'E') + let len = (buf.len() - 1) as u32; + buf[len_pos..len_pos + 4].copy_from_slice(&len.to_be_bytes()); + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +// ─── Proxy → Backend messages ──────────────────────────────────────────────── + +/// Write a StartupMessage to the backend. +pub async fn write_startup_message( + stream: &mut S, + params: &HashMap, +) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut body = Vec::new(); + body.extend_from_slice(&196608u32.to_be_bytes()); + for (key, value) in params { + body.extend_from_slice(key.as_bytes()); + body.push(0); + body.extend_from_slice(value.as_bytes()); + body.push(0); + } + body.push(0); + + let mut msg = Vec::new(); + let len: u32 = 4 + body.len() as u32; + msg.extend_from_slice(&len.to_be_bytes()); + msg.extend_from_slice(&body); + + stream.write_all(&msg).await?; + stream.flush().await?; + Ok(()) +} + +/// Write a PasswordMessage ('p') to the backend. +pub async fn write_password_message(stream: &mut S, password: &str) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'p'); + let len: u32 = 4 + password.len() as u32 + 1; + buf.extend_from_slice(&len.to_be_bytes()); + buf.extend_from_slice(password.as_bytes()); + buf.push(0); + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +/// Write a Query message ('Q'). +pub async fn write_query(stream: &mut S, sql: &str) -> Result<(), ProxyError> +where + S: AsyncWriteExt + Unpin, +{ + let mut buf = Vec::new(); + buf.push(b'Q'); + let len: u32 = 4 + sql.len() as u32 + 1; + buf.extend_from_slice(&len.to_be_bytes()); + buf.extend_from_slice(sql.as_bytes()); + buf.push(0); + stream.write_all(&buf).await?; + stream.flush().await?; + Ok(()) +} + +// ─── Backend → Proxy messages ─────────────────────────────────────────────── + +async fn read_message_length(stream: &mut S) -> Result +where + S: AsyncReadExt + Unpin, +{ + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).await?; + Ok(u32::from_be_bytes(len_buf)) +} + +/// Authentication method received from the backend. +#[derive(Debug)] +#[allow(dead_code)] +pub enum AuthMethod { + Ok, + CleartextPassword, + Md5Password { salt: [u8; 4] }, + Sasl { mechanisms: Vec }, +} + +/// Read the authentication request from the backend. +pub async fn read_authentication_method(stream: &mut S) -> Result +where + S: AsyncReadExt + Unpin, +{ + let mut type_byte = [0u8; 1]; + stream.read_exact(&mut type_byte).await?; + if type_byte[0] != b'R' { + return Err(ProxyError::ProtocolViolation(format!( + "expected AuthenticationRequest (R), got {:02x}", + type_byte[0] + ))); + } + + let len = read_message_length(stream).await?; + let mut body = vec![0u8; (len - 4) as usize]; + stream.read_exact(&mut body).await?; + + let auth_type = u32::from_be_bytes([body[0], body[1], body[2], body[3]]); + + match auth_type { + 0 => Ok(AuthMethod::Ok), + 3 => Ok(AuthMethod::CleartextPassword), + 5 => { + let mut salt = [0u8; 4]; + salt.copy_from_slice(&body[4..8]); + Ok(AuthMethod::Md5Password { salt }) + } + 10 => { + let mechanisms = std::str::from_utf8(&body[4..]) + .map_err(|_| ProxyError::ProtocolViolation("invalid SASL mechanism list".into()))? + .trim_end_matches('\0') + .split('\0') + .filter(|s| !s.is_empty()) + .map(String::from) + .collect(); + Ok(AuthMethod::Sasl { mechanisms }) + } + _ => Err(ProxyError::ProtocolViolation(format!( + "unknown auth type: {}", + auth_type + ))), + } +} + +/// Backend message types we care about during handshake drain. +#[derive(Debug)] +#[allow(dead_code)] +pub enum BackendMessage { + ReadyForQuery { + transaction_status: u8, + }, + ParameterStatus { + key: String, + value: String, + }, + BackendKeyData { + process_id: i32, + secret_key: i32, + }, + ErrorResponse { + severity: Option, + code: Option, + message: String, + }, + Unknown { + tag: u8, + }, +} + +/// Read a backend message (during handshake drain phase). +pub async fn read_backend_message(stream: &mut S) -> Result +where + S: AsyncReadExt + Unpin, +{ + let mut type_buf = [0u8; 1]; + stream.read_exact(&mut type_buf).await?; + let tag = type_buf[0]; + + let len = read_message_length(stream).await?; + let mut body = vec![0u8; (len - 4) as usize]; + stream.read_exact(&mut body).await?; + + match tag { + b'Z' => { + let status = body.first().copied().unwrap_or(b'I'); + Ok(BackendMessage::ReadyForQuery { + transaction_status: status, + }) + } + b'S' => { + let (key, rest) = split_null(&body); + let (value, _) = split_null(rest); + Ok(BackendMessage::ParameterStatus { + key: String::from_utf8_lossy(key).to_string(), + value: String::from_utf8_lossy(value).to_string(), + }) + } + b'K' => { + let process_id = i32::from_be_bytes([body[0], body[1], body[2], body[3]]); + let secret_key = i32::from_be_bytes([body[4], body[5], body[6], body[7]]); + Ok(BackendMessage::BackendKeyData { + process_id, + secret_key, + }) + } + b'E' => { + let mut severity = None; + let mut code = None; + let mut message = String::new(); + let mut i = 0; + while i < body.len() { + let field_type = body[i]; + i += 1; + if field_type == 0 { + break; + } + let rest = &body[i..]; + let (value, rest2) = split_null(rest); + i += body.len() - rest2.len(); + let value_str = String::from_utf8_lossy(value).to_string(); + match field_type { + b'S' => severity = Some(value_str), + b'C' => code = Some(value_str), + b'M' => message = value_str, + _ => {} + } + } + Ok(BackendMessage::ErrorResponse { + severity, + code, + message, + }) + } + _ => { + tracing::warn!(tag, "unknown backend message during handshake"); + Ok(BackendMessage::Unknown { tag }) + } + } +} + +fn split_null(slice: &[u8]) -> (&[u8], &[u8]) { + match slice.iter().position(|&b| b == 0) { + Some(pos) => (&slice[..pos], &slice[pos + 1..]), + None => (slice, &[][..]), + } +} diff --git a/tests/integration.rs b/tests/integration.rs new file mode 100644 index 0000000..a5440e1 --- /dev/null +++ b/tests/integration.rs @@ -0,0 +1,254 @@ +//! Integration tests for pgwire-supabase-proxy +//! +//! These tests spawn the proxy in-process against a real Postgres backend +//! (orbstack postgres-dev) and exercise it by spawning the real flicknote CLI +//! binary with a valid JWT. +//! +//! Prerequisites: +//! - kubectl context pointing at orbstack +//! - postgres-dev svc deployed in the orbstack cluster +//! - flicknote binary at ~/.cargo/bin/flicknote +//! +//! Run with (from repo root): +//! ./scripts/run-integration-tests.sh +//! cargo test --test integration -- --ignored + +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use pgwire_supabase_proxy::{serve, Claims, Config}; +use std::path::PathBuf; +use std::time::Duration; +use tokio::net::{TcpListener, TcpStream}; +use tokio::process::Command as TokioCommand; +use tokio::sync::oneshot; +use tokio::time::sleep; + +/// Postgres backend connection info — port-forward must be running on 127.0.0.1:5433. +const BACKEND_HOST: &str = "127.0.0.1"; +const BACKEND_PORT: u16 = 5433; +const BACKEND_USER: &str = "supabase_admin"; +const BACKEND_PASSWORD: &str = "dev-password"; +const BACKEND_DB: &str = "supabase"; + +/// One-time setup: patch auth.uid() once before any test runs. +/// Test JWT secret — must match what psp is configured with. +const TEST_JWT_SECRET: &str = "test-jwt-secret-for-integration-testing-only"; +/// A fixed user_id to use for all test operations. +const TEST_USER_ID: &str = "00000000-0000-0000-0000-000000000001"; + +/// Mint a JWT with the given sub claim using HMAC-SHA256. +fn mint_jwt(sub: &str) -> String { + let header = Header::new(Algorithm::HS256); + let claims = Claims { + sub: sub.to_string(), + exp: Some(9999999999), + iat: None, + role: Some("authenticated".to_string()), + email: None, + }; + encode( + &header, + &claims, + &EncodingKey::from_secret(TEST_JWT_SECRET.as_bytes()), + ) + .unwrap() +} + +/// Path to the flicknote CLI binary. +fn flicknote_path() -> PathBuf { + let home = std::env::var("HOME").unwrap_or_else(|_| "/Users/neil".to_string()); + PathBuf::from(home).join(".cargo/bin/flicknote") +} + +/// Spawn psp on an ephemeral port. +/// Returns `(port, shutdown_tx)` — caller must hold `shutdown_tx` for the +/// lifetime of the test; dropping it sends the shutdown signal. +async fn spawn_psp(database_url: String, jwt_secret: String) -> (u16, oneshot::Sender<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let port = addr.port(); + + let config = Config::new(database_url, jwt_secret, format!("127.0.0.1:{}", port)).unwrap(); + + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + + tokio::spawn(async move { + serve(config, listener, async move { + let _ = shutdown_rx.await; + }) + .await + .expect("psp server error"); + }); + + let mut attempts = 0; + loop { + attempts += 1; + if attempts > 50 { + panic!("psp server did not start in time"); + } + if TcpStream::connect(addr).await.is_ok() { + break; + } + sleep(Duration::from_millis(20)).await; + } + + sleep(Duration::from_millis(50)).await; + (port, shutdown_tx) +} + +/// Run a flicknote command, return exit status and stdout. +async fn run_flicknote(port: u16, jwt: &str, args: &[&str]) -> (bool, String) { + let db_url = format!( + "postgres://authenticated:{}@127.0.0.1:{}/supabase", + jwt, port + ); + + let mut cmd = TokioCommand::new(flicknote_path()); + cmd.env("FLICKNOTE_TOKEN", jwt) + .env("DATABASE_URL", &db_url) + .env("RUST_LOG", "warn") + .args(args); + + let output = cmd.output().await.expect("failed to spawn flicknote"); + let stdout = String::from_utf8_lossy(&output.stdout).to_string(); + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + let success = output.status.success(); + if !success && !stderr.is_empty() { + eprintln!("flicknote stderr:\n{}", stderr); + } + (success, stdout) +} + +/// Build the backend postgres URL for PSP config (URL format required by parse_backend_url). +fn psp_database_url() -> String { + format!( + "postgres://{}:{}@{}:{}/{}", + BACKEND_USER, BACKEND_PASSWORD, BACKEND_HOST, BACKEND_PORT, BACKEND_DB + ) +} + +/// Build a libpq connection string for direct tokio_postgres connections (used in ensure_setup). +fn psp_connection_string() -> String { + format!( + "host={} port={} user={} password={} dbname={}", + BACKEND_HOST, BACKEND_PORT, BACKEND_USER, BACKEND_PASSWORD, BACKEND_DB + ) +} + +/// Patches `auth.uid()` to read from `request.jwt.claim.sub`. Called at the +/// start of every test via `ensure_setup()`. `OnceCell` guarantees exactly-once +/// execution — concurrent callers wait for the first to finish. +static SETUP: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new(); + +async fn ensure_setup() { + SETUP + .get_or_init(|| async { + let url = psp_connection_string(); + let (client, connection) = tokio_postgres::connect(&url, tokio_postgres::NoTls) + .await + .expect("failed to connect to postgres for auth.uid patch"); + + tokio::spawn(async move { + if let Err(e) = connection.await { + eprintln!("setup postgres connection error: {}", e); + } + }); + + client + .batch_execute( + "CREATE OR REPLACE FUNCTION auth.uid() RETURNS uuid LANGUAGE sql STABLE AS $$ \ + SELECT nullif(current_setting('request.jwt.claim.sub', true), '')::uuid $$;", + ) + .await + .expect("failed to patch auth.uid()"); + }) + .await; +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_list() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = run_flicknote(port, &jwt, &["list"]).await; + + assert!(status, "list failed (exit != 0):\nstdout:\n{}\n", stdout); +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_list_json() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = run_flicknote(port, &jwt, &["list", "--json"]).await; + + assert!(status, "list --json failed:\nstdout:\n{}\n", stdout); + assert!( + stdout.trim().starts_with('['), + "list --json should produce a JSON array:\n{}", + stdout + ); +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_count() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = run_flicknote(port, &jwt, &["count"]).await; + + assert!(status, "note count failed:\nstdout:\n{}\n", stdout); + assert!( + stdout.trim().parse::().is_ok(), + "note count should output a number:\n{}", + stdout + ); +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_find() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = run_flicknote(port, &jwt, &["find", "test"]).await; + + assert!(status, "note find failed:\nstdout:\n{}\n", stdout); +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_project_list() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = run_flicknote(port, &jwt, &["project", "list"]).await; + + assert!(status, "note project list failed:\nstdout:\n{}\n", stdout); +} + +#[tokio::test] +#[ignore = "requires orbstack cluster — run ./scripts/run-integration-tests.sh"] +async fn integration_note_add() { + ensure_setup().await; + let psp_db_url = psp_database_url(); + let (port, _shutdown) = spawn_psp(psp_db_url, TEST_JWT_SECRET.to_string()).await; + + let jwt = mint_jwt(TEST_USER_ID); + let (status, stdout) = + run_flicknote(port, &jwt, &["add", "__psp_it__integration test note"]).await; + + assert!(status, "note add failed:\nstdout:\n{}\n", stdout); +}