diff --git a/.github/workflows/merge.yml b/.github/workflows/merge.yml index cb7817ca..42dcfc99 100644 --- a/.github/workflows/merge.yml +++ b/.github/workflows/merge.yml @@ -53,7 +53,7 @@ jobs: run: | just certs just gen_curve - cargo clippy --features notebook,lifecycle -- -D warnings + cargo clippy --features full -- -D warnings - name: Run tests run: | diff --git a/.vimspector.json b/.vimspector.json index bab1a36c..2ac19c0d 100644 --- a/.vimspector.json +++ b/.vimspector.json @@ -4,7 +4,7 @@ "adapter": "CodeLLDB", "configuration": { "request": "launch", - "program": "./target/debug/guardian" + "program": "./target/debug/bridge" } } } diff --git a/Cargo.lock b/Cargo.lock index 553ea5ba..20280a58 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -89,7 +89,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e01ed3140b2f8d422c68afa1ed2e85d996ea619c988ac834d255db32138655cb" dependencies = [ "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -226,7 +226,7 @@ dependencies = [ "actix-router", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -374,7 +374,7 @@ checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -385,7 +385,7 @@ checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -396,9 +396,9 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" [[package]] name = "autocfg" -version = "1.4.0" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "aws-lc-rs" @@ -498,7 +498,7 @@ dependencies = [ "regex", "rustc-hash 1.1.0", "shlex", - "syn 2.0.103", + "syn 2.0.104", "which", ] @@ -579,7 +579,7 @@ dependencies = [ "getrandom 0.2.16", "getrandom 0.3.3", "hex", - "indexmap 2.9.0", + "indexmap 2.10.0", "js-sys", "once_cell", "rand 0.9.1", @@ -602,9 +602,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.18.1" +version = "3.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "793db76d6187cd04dff33004d8e6c9cc4e05cd330500379d2394209271b4aeee" +checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" [[package]] name = "byteorder" @@ -629,9 +629,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.27" +version = "1.2.29" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d487aa071b5f64da6f19a3e848e3578944b726ee5a4854b82172f02aa876bfdc" +checksum = "5c1599538de2394445747c8cf7935946e3cc27e9625f889d979bfb2aaf569362" dependencies = [ "jobserver", "libc", @@ -859,9 +859,9 @@ checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" [[package]] name = "crunchy" -version = "0.2.3" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43da5946c66ffcc7745f48db692ffbb10a83bfe0afd96235c5c2a4fb23994929" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" [[package]] name = "crypto-bigint" @@ -909,7 +909,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -933,7 +933,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -944,7 +944,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -982,7 +982,7 @@ checksum = "d65d7ce8132b7c0e54497a4d9a55a1c2a0912a0d786cf894472ba818fba45762" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -993,7 +993,7 @@ checksum = "510c292c8cf384b1a340b816a9a6cf2599eb8f566a44949024af88418000c50b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1006,7 +1006,7 @@ dependencies = [ "proc-macro2", "quote", "rustc_version", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1026,7 +1026,7 @@ checksum = "bda628edc44c4bb645fbe0f758797143e4e07926f7ebf4e9bdfbd3d2ce621df3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", "unicode-xid", ] @@ -1056,7 +1056,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1118,7 +1118,7 @@ dependencies = [ "enum-ordinalize", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1169,7 +1169,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1189,7 +1189,7 @@ checksum = "0d28318a75d4aead5c4db25382e8ef717932d0346600cacae6357eb5941bc5ff" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1200,12 +1200,12 @@ checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" [[package]] name = "errno" -version = "0.3.12" +version = "0.3.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cea14ef9355e3beab063703aa9dab15afd25f0667c341310c1e5274bb1d0da18" +checksum = "778e2ac28f6c47af28e4907f13ffd1e1ddbd400980a9abd7c8df189bf578a5ad" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -1365,7 +1365,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -1507,7 +1507,7 @@ dependencies = [ "futures-sink", "futures-util", "http 0.2.12", - "indexmap 2.9.0", + "indexmap 2.10.0", "slab", "tokio", "tokio-util", @@ -1516,9 +1516,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.10" +version = "0.4.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9421a676d1b147b16b82c9225157dc629087ef8ec4d5e2960f9437a90dac0a5" +checksum = "17da50a276f1e01e0ba6c029e47b7100754904ee8a278f886546e98575380785" dependencies = [ "atomic-waker", "bytes", @@ -1526,7 +1526,7 @@ dependencies = [ "futures-core", "futures-sink", "http 1.3.1", - "indexmap 2.9.0", + "indexmap 2.10.0", "slab", "tokio", "tokio-util", @@ -1750,7 +1750,7 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "h2 0.4.10", + "h2 0.4.11", "http 1.3.1", "http-body", "httparse", @@ -1797,7 +1797,7 @@ dependencies = [ "tokio", "tokio-rustls 0.26.2", "tower-service", - "webpki-roots 1.0.0", + "webpki-roots 1.0.1", ] [[package]] @@ -1831,9 +1831,9 @@ dependencies = [ [[package]] name = "hyper-util" -version = "0.1.14" +version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc2fdfdbff08affe55bb779f33b053aa1fe5dd5b54c257343c17edfa55711bdb" +checksum = "7f66d5bd4c6f02bf0542fad85d626775bab9258cf795a4256dcaf3161114d1df" dependencies = [ "base64 0.22.1", "bytes", @@ -2027,15 +2027,26 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.9.0" +version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cea70ddb795996207ad57735b50c5982d8844f38ba9ee5f1aedcfb708a2aa11e" +checksum = "fe4cd85333e22411419a0bcae1297d25e58c9443848b11dc6a86fefe8c78a661" dependencies = [ "equivalent", "hashbrown 0.15.4", "serde", ] +[[package]] +name = "io-uring" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b86e202f00093dcba4275d4636b93ef9dd75d025ae560d2521b45ea28ab49013" +dependencies = [ + "bitflags 2.9.1", + "cfg-if", + "libc", +] + [[package]] name = "ipconfig" version = "0.3.2" @@ -2282,7 +2293,7 @@ dependencies = [ "quote", "serde", "serde_json", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2335,9 +2346,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" [[package]] name = "libc" -version = "0.2.173" +version = "0.2.174" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8cfeafaffdbc32176b64fb251369d52ea9f0a8fbc6f8759edffef7b525d64bb" +checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776" [[package]] name = "libloading" @@ -2446,7 +2457,7 @@ dependencies = [ "macro_magic_core", "macro_magic_macros", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2460,7 +2471,7 @@ dependencies = [ "macro_magic_core_macros", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2471,7 +2482,7 @@ checksum = "b02abfe41815b5bd98dbd4260173db2c116dda171dc0fe7838cb206333b83308" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2482,7 +2493,7 @@ checksum = "73ea28ee64b88876bf45277ed9a5817c1817df061a74f2b988971a12570e5869" dependencies = [ "macro_magic_core", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2555,9 +2566,9 @@ dependencies = [ [[package]] name = "mongodb" -version = "3.2.3" +version = "3.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdf4261933e5113914caec01c4bb16a7502bdaa9cf80fd87191765e7d9ff16b2" +checksum = "d0f8c69f13acf07eae386a2974f48ffd9187ea2aba8defbea9aa34e7e272c5f3" dependencies = [ "async-trait", "base64 0.13.1", @@ -2587,7 +2598,7 @@ dependencies = [ "serde", "serde_bytes", "serde_with", - "sha-1", + "sha1", "sha2", "socket2", "stringprep", @@ -2604,14 +2615,14 @@ dependencies = [ [[package]] name = "mongodb-internal-macros" -version = "3.2.3" +version = "3.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "619176c99deef0d50be51ce3193e9efd6a56ab0f4e6a38d5fd614880d148c7ae" +checksum = "b9202de265a3a8bbb43f9fe56db27c93137d4f9fb04c093f47e9c7de0c61ac7d" dependencies = [ "macro_magic", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -2804,9 +2815,9 @@ dependencies = [ [[package]] name = "openidconnect" -version = "4.0.0" +version = "4.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6dd50d4a5e7730e754f94d977efe61f611aadd3131f6a2b464f6e3a4167e8ef7" +checksum = "0d8c6709ba2ea764bbed26bce1adf3c10517113ddea6f2d4196e4851757ef2b2" dependencies = [ "base64 0.21.7", "chrono", @@ -2856,7 +2867,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3028,7 +3039,7 @@ dependencies = [ "pest_meta", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3096,7 +3107,7 @@ checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3164,12 +3175,12 @@ dependencies = [ [[package]] name = "prettyplease" -version = "0.2.34" +version = "0.2.35" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6837b9e10d61f45f987d50808f83d1ee3d206c66acf650c3e4ae2e1f6ddedf55" +checksum = "061c1221631e079b26479d25bbf2275bfe5917ae8419cd7e34f13bfc2aa7539a" dependencies = [ "proc-macro2", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3233,9 +3244,9 @@ dependencies = [ [[package]] name = "quinn-udp" -version = "0.5.12" +version = "0.5.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee4e529991f949c5e25755532370b8af5d114acae52326361d68d47af64aa842" +checksum = "fcebb1209ee276352ef14ff8732e24cc2b02bbac986cd74a4c81bcb2f9881970" dependencies = [ "cfg_aliases", "libc", @@ -3327,9 +3338,9 @@ dependencies = [ [[package]] name = "redis" -version = "0.32.0" +version = "0.32.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd673deb3d184140109c1e27b043c2354cb79399c3ef304c365fc628092ed773" +checksum = "7f0f6a8c53351d89a3869a703459995a0bcadcfa846002707fbc7e5cca235c4a" dependencies = [ "bytes", "cfg-if", @@ -3373,7 +3384,7 @@ checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3413,16 +3424,16 @@ checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" [[package]] name = "reqwest" -version = "0.12.20" +version = "0.12.22" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eabf4c97d9130e2bf606614eb937e86edac8292eaa6f422f995d7e8de1eb1813" +checksum = "cbc931937e6ca3a06e3b6c0aa7841849b160a90351d6ab467a8b9b9959767531" dependencies = [ "base64 0.22.1", "bytes", "encoding_rs", "futures-core", "futures-util", - "h2 0.4.10", + "h2 0.4.11", "http 1.3.1", "http-body", "http-body-util", @@ -3455,7 +3466,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", - "webpki-roots 1.0.0", + "webpki-roots 1.0.1", ] [[package]] @@ -3754,6 +3765,18 @@ dependencies = [ "serde_json", ] +[[package]] +name = "schemars" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82d20c4491bc164fa2f6c5d44565947a52ad80b9505d8e36f8d54c27c739fcd0" +dependencies = [ + "dyn-clone", + "ref-cast", + "serde", + "serde_json", +] + [[package]] name = "schemars_derive" version = "0.8.22" @@ -3763,7 +3786,7 @@ dependencies = [ "proc-macro2", "quote", "serde_derive_internals", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3884,7 +3907,7 @@ checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3895,7 +3918,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3904,7 +3927,7 @@ version = "1.0.140" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20068b6e96dc6c9bd23e01df8827e6c7e1f2fddd43c21810382803c136b99373" dependencies = [ - "indexmap 2.9.0", + "indexmap 2.10.0", "itoa", "memchr", "ryu", @@ -3932,9 +3955,9 @@ dependencies = [ [[package]] name = "serde_spanned" -version = "0.6.9" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +checksum = "40734c41988f7306bb04f0ecf60ec0f3f1caa34290e4e8ea471dcd3346483b83" dependencies = [ "serde", ] @@ -3953,16 +3976,17 @@ dependencies = [ [[package]] name = "serde_with" -version = "3.13.0" +version = "3.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf65a400f8f66fb7b0552869ad70157166676db75ed8181f8104ea91cf9d0b42" +checksum = "f2c45cd61fefa9db6f254525d46e392b852e0e61d9a1fd36e5bd183450a556d5" dependencies = [ "base64 0.22.1", "chrono", "hex", "indexmap 1.9.3", - "indexmap 2.9.0", + "indexmap 2.10.0", "schemars 0.9.0", + "schemars 1.0.4", "serde", "serde_derive", "serde_json", @@ -3972,14 +3996,14 @@ dependencies = [ [[package]] name = "serde_with_macros" -version = "3.13.0" +version = "3.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81679d9ed988d5e9a5e6531dc3f2c28efbd639cbd1dfb628df08edea6004da77" +checksum = "de90945e6565ce0d9a25098082ed4ee4002e047cb59892c318d66821e14bb30f" dependencies = [ "darling", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -3988,7 +4012,7 @@ version = "0.9.34+deprecated" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" dependencies = [ - "indexmap 2.9.0", + "indexmap 2.10.0", "itoa", "ryu", "serde", @@ -4005,17 +4029,6 @@ dependencies = [ "serde", ] -[[package]] -name = "sha-1" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c" -dependencies = [ - "cfg-if", - "cpufeatures", - "digest", -] - [[package]] name = "sha1" version = "0.10.6" @@ -4192,9 +4205,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.103" +version = "2.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e4307e30089d6fd6aff212f2da3a1f9e32f3223b1f010fb09b7c95f90f3ca1e8" +checksum = "17b6f705963418cdb9927482fa304bc562ece2fdd4f616084c50b7023b435a40" dependencies = [ "proc-macro2", "quote", @@ -4218,7 +4231,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4315,7 +4328,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4326,7 +4339,7 @@ checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4405,17 +4418,19 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.45.1" +version = "1.46.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75ef51a33ef1da925cea3e4eb122833cb377c61439ca401b770f54902b806779" +checksum = "0cc3a2344dafbe23a245241fe8b09735b521110d30fcefbbd5feb1797ca35d17" dependencies = [ "backtrace", "bytes", + "io-uring", "libc", "mio", "parking_lot", "pin-project-lite", "signal-hook-registry", + "slab", "socket2", "tokio-macros", "windows-sys 0.52.0", @@ -4429,7 +4444,7 @@ checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4507,44 +4522,42 @@ dependencies = [ [[package]] name = "toml" -version = "0.8.23" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +checksum = "f271e09bde39ab52250160a67e88577e0559ad77e9085de6e9051a2c4353f8f8" dependencies = [ + "indexmap 2.10.0", "serde", "serde_spanned", "toml_datetime", - "toml_edit", + "toml_parser", + "toml_writer", + "winnow", ] [[package]] name = "toml_datetime" -version = "0.6.11" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +checksum = "bade1c3e902f58d73d3f294cd7f20391c1cb2fbcb643b73566bc773971df91e3" dependencies = [ "serde", ] [[package]] -name = "toml_edit" -version = "0.22.27" +name = "toml_parser" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +checksum = "b5c1c469eda89749d2230d8156a5969a69ffe0d6d01200581cdc6110674d293e" dependencies = [ - "indexmap 2.9.0", - "serde", - "serde_spanned", - "toml_datetime", - "toml_write", "winnow", ] [[package]] -name = "toml_write" -version = "0.1.2" +name = "toml_writer" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" +checksum = "b679217f2848de74cabd3e8fc5e6d66f40b7da40f8e1954d92054d9010690fd5" [[package]] name = "tower" @@ -4616,7 +4629,7 @@ checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4845,7 +4858,7 @@ name = "utils" version = "0.1.0" dependencies = [ "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -4940,7 +4953,7 @@ dependencies = [ "log", "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", "wasm-bindgen-shared", ] @@ -4975,7 +4988,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -5030,9 +5043,9 @@ checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" [[package]] name = "webpki-roots" -version = "1.0.0" +version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2853738d1cc4f2da3a225c18ec6c3721abb31961096e9dbf5ab35fa88b19cfdb" +checksum = "8782dd5a41a24eed3a4f40b606249b3e236ca61adf1f25ea4d45c73de122b502" dependencies = [ "rustls-pki-types", ] @@ -5107,7 +5120,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -5118,7 +5131,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -5129,9 +5142,9 @@ checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" [[package]] name = "windows-registry" -version = "0.5.2" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b3bab093bdd303a1240bb99b8aba8ea8a69ee19d34c9e2ef9594e708a4878820" +checksum = "5b8a9ed28765efc97bbc954883f4e6796c33a06546ebafacbabee9696967499e" dependencies = [ "windows-link", "windows-result", @@ -5183,6 +5196,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.2", +] + [[package]] name = "windows-targets" version = "0.48.5" @@ -5373,9 +5395,6 @@ name = "winnow" version = "0.7.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74c7b26e3480b707944fc872477815d29a8e429d2f93a1ce000f5fa84a15cbcd" -dependencies = [ - "memchr", -] [[package]] name = "winreg" @@ -5440,28 +5459,28 @@ checksum = "38da3c9736e16c5d3c8c597a9aaa5d1fa565d0532ae05e27c24aa62fb32c0ab6" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", "synstructure", ] [[package]] name = "zerocopy" -version = "0.8.25" +version = "0.8.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1702d9583232ddb9174e01bb7c15a2ab8fb1bc6f227aa1233858c351a3ba0cb" +checksum = "1039dd0d3c310cf05de012d8a39ff557cb0d23087fd44cad61df08fc31907a2f" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.25" +version = "0.8.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28a6e20d751156648aa063f3800b706ee209a32c0b4d9f24be3d980b01be55ef" +checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -5481,7 +5500,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", "synstructure", ] @@ -5502,7 +5521,7 @@ checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] @@ -5535,7 +5554,7 @@ checksum = "5b96237efa0c878c64bd89c436f661be4e46b2f3eff1ebb976f7ef2321d2f58f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.103", + "syn 2.0.104", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 271743a0..6bd0049c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,11 +11,13 @@ path = "./src/main.rs" members = ["utils"] [features] -notebook = ["dep:kube", "dep:k8s-openapi", "dep:schemars", "dep:either"] +kubernetes = ["dep:kube", "dep:k8s-openapi", "dep:schemars", "dep:either"] +notebook = ["kubernetes"] +openwebui = ["kubernetes"] lifecycle = ["notebook"] observe = [] mcp = [] -full = ["notebook", "lifecycle", "observe", "mcp"] +full = ["notebook", "lifecycle", "observe", "mcp", "openwebui"] [dependencies] # Async dep @@ -37,7 +39,7 @@ actix-files = "0.6.6" tera = { version = "1.20.0", features= ["builtins"] } reqwest = { version = "0.12", features = ["stream", "json"] } url = "2.5.0" -toml = "0.8.14" +toml = "0.9.0" urlencoding = "2.1.3" actix-ws = "0.3.0" tokio-tungstenite = { version = "0.27", features = ["url", "rustls-tls-native-roots"] } @@ -60,7 +62,7 @@ regex = "~1.11" # Error handling thiserror = "2" # DB -mongodb = "~3.2" +mongodb = "=3.2" redis = { version = "0.32", features = ["tokio-comp", "num-bigint"] } #macro utils = { path = "utils" } @@ -89,9 +91,11 @@ debug = 2 overflow-checks = true incremental = true lto = false +panic = "unwind" [profile.release] opt-level = 3 overflow-checks = false debug-assertions = false -incremental = false +incremental = true lto = true +panic = "abort" diff --git a/Dockerfile b/Dockerfile index 239b44a6..36818acb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # Stage 1 build -FROM rust:1.87.0 AS builder +FROM rust:1.88.0 AS builder WORKDIR /app @@ -9,6 +9,7 @@ ARG NOTEBOOK=false ARG LIFECYCLE=false ARG OBSERVE=false ARG MCP=false +ARG OWUI=false RUN < Vec { diff --git a/src/auth/openid.rs b/src/auth/openid.rs index 60cc846a..87c1cae0 100644 --- a/src/auth/openid.rs +++ b/src/auth/openid.rs @@ -187,6 +187,6 @@ mod tests { init_once().await; let openid = OPENID_W3.get().unwrap(); let (u, c, n) = openid.get_client_resources(); - println!("{:?} {:?} {:?}", u, c, n); + println!("{u:?} {c:?} {n:?}"); } } diff --git a/src/config/mod.rs b/src/config/mod.rs index b4118fa6..f326a71a 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -30,13 +30,22 @@ pub struct Configuration { pub argon_config: argon2::Config<'static>, pub db: Database, pub cache: CacheDB, + #[cfg(feature = "notebook")] pub notebooks: HashMap, + #[cfg(feature = "notebook")] pub notebook_namespace: String, pub app_name: String, pub app_discription: String, pub company: String, pub oidc: HashMap, pub observability_cred: Option<(String, String)>, + #[cfg(feature = "openwebui")] + pub owui_namespace: String, + #[cfg(feature = "openwebui")] + pub openweb_url: String, + #[cfg(feature = "openwebui")] + pub moleviewer_url: String, + pub bridge_url: String, } pub struct Database { @@ -117,12 +126,22 @@ pub fn init_once() -> Configuration { let public_key = fs::read("certs/public-key.pem").unwrap(); let decoder = DecodingKey::from_ec_pem(&public_key).unwrap(); + // JWK + let mut hasher = sha2::Sha256::new(); + hasher.update(&public_key); + let kid = BASE64_URL_SAFE_NO_PAD.encode(hasher.finalize()); + let key = p256::PublicKey::from_public_key_pem(&String::from_utf8_lossy(&public_key)).unwrap(); + let jwk = JWK::new(key.to_jwk(), kid.clone()); + let mut validation = Validation::new(Algorithm::ES256); validation.set_audience(&AUD); validation.leeway = 0; let (config_location_str, database_location_str) = if cfg!(debug_assertions) { - ("config/configurations_sample.toml", "config/database_sample.toml") + ( + "config/configurations_sample.toml", + "config/database_sample.toml", + ) } else { ("config/configurations.toml", "config/database.toml") }; @@ -130,12 +149,6 @@ pub fn init_once() -> Configuration { let conf_table: toml::Table = toml::from_str(&read_to_string(PathBuf::from_str(config_location_str).unwrap()).unwrap()) .unwrap(); - let mut hasher = sha2::Sha256::new(); - hasher.update(&public_key); - let kid = BASE64_URL_SAFE_NO_PAD.encode(hasher.finalize()); - - let key = p256::PublicKey::from_public_key_pem(&String::from_utf8_lossy(&public_key)).unwrap(); - let jwk = JWK::new(key.to_jwk(), kid.clone()); let db_table: toml::Table = toml::from_str(&read_to_string(PathBuf::from_str(database_location_str).unwrap()).unwrap()) @@ -187,7 +200,6 @@ pub fn init_once() -> Configuration { let app_name = app_conf["name"].as_str().unwrap().to_string(); let app_discription = app_conf["description"].as_str().unwrap().to_string(); let company = app_conf["company"].as_str().unwrap().to_string(); - let notebook_namespace = app_conf["notebook_namespace"].as_str().unwrap().to_string(); let observability_cred = match ( app_conf["observability_access_token"] @@ -201,11 +213,25 @@ pub fn init_once() -> Configuration { _ => None, }; + #[cfg(feature = "notebook")] + let notebook_namespace = app_conf["notebook_namespace"].as_str().unwrap().to_string(); + #[cfg(feature = "notebook")] let notebooks: HashMap = toml::from_str( &read_to_string(PathBuf::from_str("config/notebook.toml").unwrap()).unwrap(), ) .unwrap(); + #[cfg(feature = "openwebui")] + let (owui_namespace, openweb_url, moleviewer_url) = { + ( + app_conf["owui_namespace"].as_str().unwrap().to_string(), + app_conf["openweb_url"].as_str().unwrap().to_string(), + app_conf["moleviewer_url"].as_str().unwrap().to_string(), + ) + }; + + let bridge_url = app_conf["bridge_url"].as_str().unwrap().to_string(); + Configuration { encoder, decoder, @@ -215,13 +241,22 @@ pub fn init_once() -> Configuration { argon_config: argon2::Config::default(), db, cache, + #[cfg(feature = "notebook")] notebooks, + #[cfg(feature = "notebook")] notebook_namespace, app_name, app_discription, company, oidc: oidc_map, observability_cred, + #[cfg(feature = "openwebui")] + owui_namespace, + #[cfg(feature = "openwebui")] + openweb_url, + #[cfg(feature = "openwebui")] + moleviewer_url, + bridge_url, } } @@ -235,13 +270,18 @@ mod tests { #[test] fn test_config_init_once() { let config = init_once(); - let workbench = config.notebooks.get("open_ad_workbench").unwrap(); - assert_eq!(workbench.pull_policy, "Always"); - assert_eq!(workbench.working_dir, Some("/opt/app-root/src".to_string())); - assert_eq!( - workbench.start_up_url, - Some("lab/tree/start_menu.ipynb".to_string()) - ); + #[cfg(feature = "notebook")] + { + let workbench = config.notebooks.get("open_ad_workbench").unwrap(); + assert_eq!(workbench.pull_policy, "Always"); + assert_eq!(workbench.working_dir, Some("/opt/app-root/src".to_string())); + assert_eq!( + workbench.start_up_url, + Some("lab/tree/start_menu.ipynb".to_string()) + ); + } + + assert_eq!(config.app_name, "Open Accelerated Discovery"); } #[test] diff --git a/src/db/keydb.rs b/src/db/keydb.rs index 2470f862..04a6a361 100644 --- a/src/db/keydb.rs +++ b/src/db/keydb.rs @@ -44,26 +44,20 @@ impl CacheDB { let sub = sub.as_ref(); let _: () = self .get_connection() - .set_ex(format!("session:{}", sub), session_id, expiration) + .set_ex(format!("session:{sub}"), session_id, expiration) .await?; Ok(()) } pub async fn get_session_id>(&self, sub: T) -> Result { let sub = sub.as_ref(); - let session_id: String = self - .get_connection() - .get(format!("session:{}", sub)) - .await?; + let session_id: String = self.get_connection().get(format!("session:{sub}")).await?; Ok(session_id) } pub async fn del_session_id>(&self, sub: T) -> Result<()> { let sub = sub.as_ref(); - let _: () = self - .get_connection() - .del(format!("session:{}", sub)) - .await?; + let _: () = self.get_connection().del(format!("session:{sub}")).await?; Ok(()) } } @@ -109,10 +103,10 @@ mod tests { let data_out: String = conn.get("test").await.unwrap(); let data_out_value: serde_json::Value = serde_json::from_str(&data_out).unwrap(); assert_eq!(data_in, data_out_value); - println!("data_out: {}", data_out); + println!("data_out: {data_out}"); let not_exist: Option = conn.get("nothing").await.unwrap(); - println!("{:?}", not_exist); + println!("{not_exist:?}"); // let mut stream = cache.get_async_sub("maintenance").await.unwrap(); // let msg = stream.next().await.unwrap(); diff --git a/src/db/models.rs b/src/db/models.rs index 6f03b076..12b25fbb 100644 --- a/src/db/models.rs +++ b/src/db/models.rs @@ -184,6 +184,18 @@ pub struct AppPayload { pub password: String, } +pub static OBSERVE: &str = "observe"; +#[derive(Debug, Deserialize, Serialize)] +pub struct ObserveEventEntry { + pub sub: String, + // pub group: String, + pub property: String, + #[serde(rename = "requestDate")] + pub request_date: time::OffsetDateTime, + #[serde(rename = "expireSoonAfter")] + pub expire_soon_after: bson::DateTime, +} + /// This is the form verison of the Group struct #[derive(Debug)] pub struct GroupForm { @@ -217,6 +229,11 @@ pub struct BridgeCookie { pub session_id: Option, } +#[derive(Debug, Deserialize, Serialize, Clone)] +pub struct OWUICookie { + pub subject: String, +} + #[derive(Debug, Deserialize, Serialize, Clone)] pub struct Config { pub notebook_persist_pvc: Option, diff --git a/src/db/mongo.rs b/src/db/mongo.rs index 47f8f25d..699d6eac 100644 --- a/src/db/mongo.rs +++ b/src/db/mongo.rs @@ -15,13 +15,13 @@ use serde::{Serialize, de::DeserializeOwned}; use crate::{ config::CONFIG, - db::models::Apps, + db::models::{Apps, ObserveEventEntry}, errors::{BridgeError, Result}, }; use super::{ Database, - models::{APPS, GROUP, Group, GroupSubs, LOCKS, Locks, USER, User}, + models::{APPS, GROUP, Group, GroupSubs, LOCKS, Locks, OBSERVE, USER, User}, }; type Pipeline = Vec; @@ -48,7 +48,7 @@ impl ObjectID { pub static DBCONN: OnceLock = OnceLock::new(); pub static DBNAME: LazyLock<&str> = LazyLock::new(|| &CONFIG.db.name); -static COLLECTIONS: [&str; 4] = [USER, GROUP, LOCKS, APPS]; +static COLLECTIONS: [&str; 5] = [USER, GROUP, LOCKS, APPS, OBSERVE]; impl DB { pub async fn init_once(database: &'static str) -> Result<()> { @@ -71,18 +71,46 @@ impl DB { .unique(true) .build() } + + fn not_unique(f: impl Into + ToString) -> IndexOptions { + IndexOptions::builder() + .name(Some(f.to_string())) + .unique(false) + .build() + } + Self::create_index::(&dbs, USER, "email", 1, unique).await?; + Self::create_index::(&dbs, GROUP, "name", "text", unique).await?; - Self::create_index::(&dbs, LOCKS, "expireSoonAfter", 1, |f| { + + Self::create_index::(&dbs, LOCKS, "expireSoonAfter", 1, |f: &str| { IndexOptions::builder() .name(Some(f.to_string())) + // this means number of seconds after the time set .expire_after(Duration::from_secs(0)) // 1 hour .build() }) .await?; Self::create_index::(&dbs, LOCKS, "leaseName", "text", unique).await?; + Self::create_index::(&dbs, APPS, "client_id", "text", unique).await?; + Self::create_index::(&dbs, OBSERVE, "sub", 1, not_unique).await?; + Self::create_index::(&dbs, OBSERVE, "group", 1, not_unique).await?; + Self::create_index::( + &dbs, + OBSERVE, + "expireSoonAfter", + 1, + |f: &str| { + IndexOptions::builder() + .name(Some(f.to_string())) + .expire_after(Duration::from_secs(0)) // 6 months + .build() + }, + ) + .await?; + DBCONN.get_or_init(|| dbs); Ok(()) @@ -179,6 +207,15 @@ impl DB { } } +pub mod helper { + use mongodb::bson; + + pub fn i64_to_bson_datatime(sec: i64) -> bson::DateTime { + let time_in_ms = sec * 1_000; + bson::DateTime::from_millis(time_in_ms) + } +} + // pub trait Database<'c, R1 = User, Q = Document, N = &'c str, C = &'c str, R2 = Bson, R3 = u64> { impl Database for DB where diff --git a/src/errors/mod.rs b/src/errors/mod.rs index c2ba121f..b56e665e 100644 --- a/src/errors/mod.rs +++ b/src/errors/mod.rs @@ -65,18 +65,17 @@ pub enum BridgeError { RecordSearchError(String), #[error("{0}")] WSError(#[from] actix_web::error::Error), - #[cfg(feature = "notebook")] + #[cfg(feature = "kubernetes")] #[error("{0}")] KubeError(#[from] kube::Error), - #[cfg(feature = "notebook")] + #[cfg(feature = "kubernetes")] #[error("{0}")] - NotebookExistsError(String), + CRDExistsError(String), #[cfg(feature = "notebook")] #[error("{0}")] NotebookAccessError(String), - #[cfg(feature = "notebook")] + #[cfg(feature = "kubernetes")] #[error("{0}")] - #[cfg(feature = "notebook")] KubeClientError(String), #[error("{0}")] ReqwestError(#[from] reqwest::Error), @@ -173,9 +172,9 @@ impl ResponseError for BridgeError { BridgeError::OpenIDError(_) => StatusCode::INTERNAL_SERVER_ERROR, BridgeError::Argon2Error(_) => StatusCode::INTERNAL_SERVER_ERROR, - #[cfg(feature = "notebook")] + #[cfg(feature = "kubernetes")] BridgeError::KubeError(_) => StatusCode::INTERNAL_SERVER_ERROR, - #[cfg(feature = "notebook")] + #[cfg(feature = "kubernetes")] BridgeError::KubeClientError(_) => StatusCode::INTERNAL_SERVER_ERROR, BridgeError::Unauthorized(_) => StatusCode::UNAUTHORIZED, @@ -189,8 +188,8 @@ impl ResponseError for BridgeError { #[cfg(feature = "notebook")] BridgeError::NotebookAccessError(_) => StatusCode::FORBIDDEN, - #[cfg(feature = "notebook")] - BridgeError::NotebookExistsError(_) => StatusCode::CONFLICT, + #[cfg(feature = "kubernetes")] + BridgeError::CRDExistsError(_) => StatusCode::CONFLICT, } } } diff --git a/src/kube/mod.rs b/src/kube/mod.rs index f8c94e92..1e863a0b 100644 --- a/src/kube/mod.rs +++ b/src/kube/mod.rs @@ -1,143 +1,15 @@ -use std::{fmt::Debug, sync::OnceLock}; - -use k8s_openapi::{ - NamespaceResourceScope, - api::core::v1::{Namespace, PersistentVolumeClaim, Pod}, -}; -use kube::{ - Api, Client, Resource, - api::{DeleteParams, ObjectMeta, PostParams}, -}; -use reqwest::StatusCode; -use serde::{Deserialize, Serialize}; -use tracing::info; - -use crate::errors::{BridgeError, Result}; - mod models; -pub use models::{NAMESPACE, Notebook, NotebookSpec, PVCSpec}; - -mod notebook_lifecycle; -pub use notebook_lifecycle::{LifecycleStream, Medium, notebook_lifecycle}; - -pub struct KubeAPI { - model: M, -} - -static KUBECLIENT: OnceLock = OnceLock::new(); - -pub async fn init_once() { - // ok to fail since we should not start if we can't connect to k8s - let client = Client::try_default() - .await - .expect("Failed to connect to k8s"); - info!("Connected to k8s"); - KUBECLIENT.get_or_init(|| client); -} - -impl KubeAPI -where - M: Resource - + Clone - + Debug - + for<'a> Deserialize<'a> // hrtb - + Serialize, - ::DynamicType: Default, -{ - pub fn new(model: M) -> Self { - Self { model } - } - - pub fn get_kube_client() -> Result<&'static Client> { - KUBECLIENT.get().ok_or(BridgeError::KubeClientError( - "Could not get kube client".to_string(), - )) - } - - pub async fn create(&self) -> Result { - let crd = Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let pp = PostParams::default(); - let res = match crd.create(&pp, &self.model).await { - Ok(res) => res, - Err(e) => match e { - kube::Error::Api(ref error_response) => { - if error_response.reason == "AlreadyExists" { - return Err(BridgeError::NotebookExistsError( - "AlreadyExists".to_string(), - )); - } - Err(e)? - } - _ => Err(e)?, - }, - }; - Ok(res) - } - - pub async fn delete(name: &str) -> Result { - let crd = Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let dp = DeleteParams::default(); - let status = match crd.delete(name, &dp).await? { - // resource is in the process of being deleted - either::Either::Left(_) => StatusCode::PROCESSING, - // resource has been deleted - either::Either::Right(_) => StatusCode::OK, - }; - Ok(status) - } - - /// Create a namespace if it does not exist. Returns `None` if the namespace already exists. - /// Returns `Some(())` if the namespace was created. - pub async fn make_namespace(name: &str) -> Result> { - let ns = Api::::all(Self::get_kube_client()?.clone()); - if ns.get_opt(name).await?.is_some() { - return Ok(None); - } - - let new_ns = Namespace { - metadata: ObjectMeta { - name: Some(name.to_string()), - ..Default::default() - }, - spec: Default::default(), - status: Default::default(), - }; - - ns.create(&PostParams::default(), &new_ns).await?; - Ok(Some(())) - } - - pub async fn check_pod_running(name: &str) -> Result { - let pods = Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let pod = pods.get(name).await?; - Ok(pod - .status - .as_ref() - .map(|status| status.phase == Some("Running".to_string())) - .unwrap_or(false)) - } - - pub async fn check_pvc_exists(name: &str) -> Result { - let pvcs = - Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let pvc = pvcs.get(name).await; - Ok(pvc.is_ok()) - } - - pub async fn get_all_pods() -> Result> { - let pods = Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let list = pods.list(&Default::default()).await?; - Ok(list.items) - } +pub use models::*; + +#[cfg(feature = "notebook")] +mod notebook; +#[cfg(feature = "notebook")] +pub use notebook::{ + LifecycleStream, Medium, NOTEBOOK_NAMESPACE, Notebook, NotebookSpec, PVCSpec, + notebook_lifecycle, +}; - pub async fn get_pod_ip(name: &str) -> Result { - let pods = Api::::namespaced(Self::get_kube_client()?.clone(), *NAMESPACE); - let pod = pods.get(name).await?; - Ok(pod - .status - .as_ref() - .and_then(|status| status.pod_ip.as_deref()) - .unwrap_or_default() - .to_string()) - } -} +#[cfg(feature = "openwebui")] +mod openwebui; +#[cfg(feature = "openwebui")] +pub use openwebui::OWUI; diff --git a/src/kube/models.rs b/src/kube/models.rs index 945ad6dc..16e989b1 100644 --- a/src/kube/models.rs +++ b/src/kube/models.rs @@ -1,320 +1,135 @@ -use std::{collections::BTreeMap, sync::LazyLock}; +use std::{fmt::Debug, sync::OnceLock}; use k8s_openapi::{ - api::core::v1::{PersistentVolumeClaim, VolumeResourceRequirements}, - apimachinery::pkg::api::resource::Quantity, + NamespaceResourceScope, + api::core::v1::{Namespace, PersistentVolumeClaim, Pod}, }; -use kube::{CustomResource, api::ObjectMeta}; -use schemars::JsonSchema; +use kube::{ + Api, Client, Resource, + api::{DeleteParams, ObjectMeta, PostParams}, +}; +use reqwest::StatusCode; use serde::{Deserialize, Serialize}; +use tracing::info; -use crate::config::CONFIG; - -pub static NAMESPACE: LazyLock<&str> = LazyLock::new(|| &CONFIG.notebook_namespace); +use crate::errors::{BridgeError, Result}; -// Define the Notebook CRD struct -#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)] -#[kube(group = "kubeflow.org", version = "v1", kind = "Notebook", namespaced)] -pub struct NotebookSpec { - template: NotebookTemplateSpec, +pub struct KubeAPI { + model: M, } -impl NotebookSpec { - pub fn new( - name: String, - notebook_image_name: &str, - volume_name: String, - notebook_start_url: &mut Option, - max_idle_time: &mut Option, - env_to_add: Vec<(String, String)>, - ) -> Self { - let notebook_image = CONFIG.notebooks.get(notebook_image_name).unwrap(); - let mut notebook_env = notebook_image.notebook_env.clone().unwrap_or_default(); - - notebook_env.push(format!( - "--ServerApp.base_url='notebook/{}/{}'", - *NAMESPACE, name - )); - - let mut env = vec![EnvVar { - name: "NOTEBOOK_ARGS".to_string(), - value: notebook_env.join(" "), - }]; +static KUBECLIENT: OnceLock = OnceLock::new(); - if !env_to_add.is_empty() { - env_to_add.into_iter().for_each(|(name, value)| { - env.push(EnvVar { name, value }); - }); - } +pub async fn init_once() { + // ok to fail since we should not start if we can't connect to k8s + let client = Client::try_default() + .await + .expect("Failed to connect to k8s"); + info!("Connected to k8s"); + KUBECLIENT.get_or_init(|| client); +} - // TODO: look into removing this clone - if let Some(envs) = notebook_image.env.clone() { - envs.into_iter().for_each(|(name, value)| { - env.push(EnvVar { name, value }); - }); - } +impl KubeAPI +where + M: Resource + + Clone + + Debug + + for<'a> Deserialize<'a> // hrtb + + Serialize, + ::DynamicType: Default, +{ + pub fn new(model: M) -> Self { + Self { model } + } - *notebook_start_url = notebook_image.start_up_url.clone(); - *max_idle_time = notebook_image.max_idle_time; + pub fn get_kube_client() -> Result<&'static Client> { + KUBECLIENT.get().ok_or(BridgeError::KubeClientError( + "Could not get kube client".to_string(), + )) + } - NotebookSpec { - template: NotebookTemplateSpec { - spec: PodSpec { - containers: vec![ContainerSpec { - name, - image: notebook_image.url.clone(), - resources: Some(ResourceRequirements { - requests: BTreeMap::from([ - ("cpu".to_string(), "2".to_string()), - ("memory".to_string(), "4Gi".to_string()), - ]), - limits: BTreeMap::from([ - ("cpu".to_string(), "2".to_string()), - ("memory".to_string(), "4Gi".to_string()), - ]), - }), - image_pull_policy: notebook_image.pull_policy.clone(), - volume_mounts: Some(vec![VolumeMount { - name: volume_name.clone(), - mount_path: notebook_image.volume_mnt_path.clone().unwrap_or_default(), - }]), - command: notebook_image.command.clone(), - args: notebook_image.args.clone(), - workingdir: notebook_image.working_dir.clone(), - env: Some(env), - }], - image_pull_secrets: notebook_image - .secret - .clone() - .map(|secret| vec![ImagePullSecret { name: secret }]), - volumes: Some(vec![VolumeSpec { - name: volume_name.clone(), - persistent_volume_claim: Some(PersistentVolumeClaimSpec { - claim_name: volume_name, - read_only: None, - }), - config_map: None, - }]), - }, + pub async fn create(&self, namespace: &str) -> Result { + let crd = Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let pp = PostParams::default(); + let res = match crd.create(&pp, &self.model).await { + Ok(res) => res, + Err(e) => match e { + kube::Error::Api(ref error_response) => { + if error_response.reason == "AlreadyExists" { + return Err(BridgeError::CRDExistsError("AlreadyExists".to_string())); + } + Err(e)? + } + _ => Err(e)?, }, - } + }; + Ok(res) } -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct NotebookTemplateSpec { - spec: PodSpec, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct PodSpec { - containers: Vec, - #[serde(rename = "imagePullSecrets")] - image_pull_secrets: Option>, - volumes: Option>, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct ImagePullSecret { - name: String, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct ContainerSpec { - name: String, - image: String, - resources: Option, - #[serde(rename = "workingDir")] - workingdir: Option, - #[serde(rename = "imagePullPolicy")] - image_pull_policy: String, - #[serde(rename = "volumeMounts")] - volume_mounts: Option>, - command: Option>, - args: Option>, - env: Option>, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -struct ResourceRequirements { - requests: BTreeMap, - limits: BTreeMap, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct EnvVar { - name: String, - value: String, -} -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct VolumeMount { - name: String, - #[serde(rename = "mountPath")] - mount_path: String, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct VolumeSpec { - name: String, - #[serde(rename = "persistentVolumeClaim")] - #[serde(skip_serializing_if = "Option::is_none")] - persistent_volume_claim: Option, - #[serde(rename = "configMap")] - #[serde(skip_serializing_if = "Option::is_none")] - config_map: Option, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct ConfigMapSpec { - pub name: String, -} - -#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] -pub struct PersistentVolumeClaimSpec { - #[serde(rename = "claimName")] - claim_name: String, - #[serde(rename = "readOnly")] - #[serde(skip_serializing_if = "Option::is_none")] - read_only: Option, -} + pub async fn delete(name: &str, namespace: &str) -> Result { + let crd = Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let dp = DeleteParams::default(); + let status = match crd.delete(name, &dp).await? { + // resource is in the process of being deleted + either::Either::Left(_) => StatusCode::PROCESSING, + // resource has been deleted + either::Either::Right(_) => StatusCode::OK, + }; + Ok(status) + } -// Define the PVC Spec -pub struct PVCSpec { - pub spec: PersistentVolumeClaim, -} + /// Create a namespace if it does not exist. Returns `None` if the namespace already exists. + /// Returns `Some(())` if the namespace was created. + pub async fn make_namespace(name: &str) -> Result> { + let ns = Api::::all(Self::get_kube_client()?.clone()); + if ns.get_opt(name).await?.is_some() { + return Ok(None); + } -impl PVCSpec { - pub fn new(name: String, storage_size: usize) -> Self { - Self { - spec: PersistentVolumeClaim { - metadata: ObjectMeta { - name: Some(name), - ..Default::default() - }, - spec: Some(k8s_openapi::api::core::v1::PersistentVolumeClaimSpec { - access_modes: Some(vec!["ReadWriteOnce".to_string()]), - resources: Some(VolumeResourceRequirements { - requests: Some(BTreeMap::from([( - "storage".to_string(), - Quantity(storage_size.to_string() + "Gi"), - )])), - ..Default::default() - }), - ..Default::default() - }), + let new_ns = Namespace { + metadata: ObjectMeta { + name: Some(name.to_string()), ..Default::default() }, - } - } -} - -#[cfg(test)] -mod test { - use super::*; - use serde_json::json; - - #[test] - fn test_notebook_spec() { - let name = "notebook".to_string(); - let volume_name = "notebook-volume".to_string(); - let mut start_url = None; - let mut max_idle_time = None; - - let spec = NotebookSpec::new( - name, - "open_ad_workbench", - volume_name, - &mut start_url, - &mut max_idle_time, - vec![], - ); - - // TODO: Remove this test or refactor, so it doesn't break all the time the notebok config - // is updated - let expected = json!({ - "template": { - "spec": { - "containers": [{ - "name": "notebook", - "image": "quay.io/ibmdpdev/openad_workbench_prod:latest", - "resources": { - "requests": { - "cpu": "2", - "memory": "4Gi" - }, - "limits": { - "cpu": "2", - "memory": "4Gi" - } - }, - "workingDir": "/opt/app-root/src", - "imagePullPolicy": "Always", - "volumeMounts": [ - { - "name": "notebook-volume", - "mountPath": "/opt/app-root/src" - } - ], - "command": null, - "args": null, - "env": [ - { - "name": "NOTEBOOK_ARGS", - "value": "--ServerApp.token='' --ServerApp.password='' --ServerApp.notebook_dir='/opt/app-root/src' --ServerApp.quit_button=False --LabApp.default_url='/lab/tree/start_menu.ipynb' --ServerApp.default_url='/lab/tree/start_menu.ipynb' --ServerApp.trust_xheaders=True --ServerApp.base_url='notebook/notebook/notebook'" - }, - { - - "name":"PROXY_URL", - "value": "https://open.accelerate.science/proxy" - } - ], - }], - "imagePullSecrets": [ - { - "name": "ibmdpdev-openad-pull-secret" - } - ], - "volumes": [{ - "name": "notebook-volume", - "persistentVolumeClaim": { - "claimName": "notebook-volume" - } - }] - }, - } - }); + spec: Default::default(), + status: Default::default(), + }; - let actual = serde_json::to_value(&spec).unwrap(); - assert_eq!(actual, expected); - assert_eq!(start_url, Some("lab/tree/start_menu.ipynb".to_string())); - assert_eq!(max_idle_time, Some(86400)); + ns.create(&PostParams::default(), &new_ns).await?; + Ok(Some(())) } - #[test] - fn test_pvc_spec() { - let name = "notebook-volume".to_string(); - let storage_size = 10; + pub async fn check_pod_running(name: &str, namespace: &str) -> Result { + let pods = Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let pod = pods.get(name).await?; + Ok(pod + .status + .as_ref() + .map(|status| status.phase == Some("Running".to_string())) + .unwrap_or(false)) + } - let spec = PVCSpec::new(name, storage_size); + pub async fn check_pvc_exists(name: &str, namespace: &str) -> Result { + let pvcs = + Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let pvc = pvcs.get(name).await; + Ok(pvc.is_ok()) + } - let expected = json!({ - "apiVersion": "v1", - "kind": "PersistentVolumeClaim", - "metadata": { - "name": "notebook-volume" - }, - "spec": { - "accessModes": ["ReadWriteOnce"], - "resources": { - "requests": { - "storage": "10Gi" - } - } - } - }); + pub async fn get_all_pods(namespace: &str) -> Result> { + let pods = Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let list = pods.list(&Default::default()).await?; + Ok(list.items) + } - let actual = serde_json::to_value(&spec.spec).unwrap(); - assert_eq!(actual, expected); + pub async fn get_pod_ip(name: &str, namespace: &str) -> Result { + let pods = Api::::namespaced(Self::get_kube_client()?.clone(), namespace); + let pod = pods.get(name).await?; + Ok(pod + .status + .as_ref() + .and_then(|status| status.pod_ip.as_deref()) + .unwrap_or_default() + .to_string()) } } diff --git a/src/kube/notebook/mod.rs b/src/kube/notebook/mod.rs new file mode 100644 index 00000000..29ccf39f --- /dev/null +++ b/src/kube/notebook/mod.rs @@ -0,0 +1,323 @@ +use std::{collections::BTreeMap, sync::LazyLock}; + +use k8s_openapi::{ + api::core::v1::{PersistentVolumeClaim, VolumeResourceRequirements}, + apimachinery::pkg::api::resource::Quantity, +}; +use kube::{CustomResource, api::ObjectMeta}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::config::CONFIG; + +pub static NOTEBOOK_NAMESPACE: LazyLock<&str> = LazyLock::new(|| &CONFIG.notebook_namespace); + +mod notebook_lifecycle; +pub use notebook_lifecycle::{LifecycleStream, Medium, notebook_lifecycle}; + +// Define the Notebook CRD struct +#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)] +#[kube(group = "kubeflow.org", version = "v1", kind = "Notebook", namespaced)] +pub struct NotebookSpec { + template: NotebookTemplateSpec, +} + +impl NotebookSpec { + pub fn new( + name: String, + notebook_image_name: &str, + volume_name: String, + notebook_start_url: &mut Option, + max_idle_time: &mut Option, + env_to_add: Vec<(String, String)>, + ) -> Self { + let notebook_image = CONFIG.notebooks.get(notebook_image_name).unwrap(); + let mut notebook_env = notebook_image.notebook_env.clone().unwrap_or_default(); + + notebook_env.push(format!( + "--ServerApp.base_url='notebook/{}/{}'", + *NOTEBOOK_NAMESPACE, name + )); + + let mut env = vec![EnvVar { + name: "NOTEBOOK_ARGS".to_string(), + value: notebook_env.join(" "), + }]; + + if !env_to_add.is_empty() { + env_to_add.into_iter().for_each(|(name, value)| { + env.push(EnvVar { name, value }); + }); + } + + // TODO: look into removing this clone + if let Some(envs) = notebook_image.env.clone() { + envs.into_iter().for_each(|(name, value)| { + env.push(EnvVar { name, value }); + }); + } + + *notebook_start_url = notebook_image.start_up_url.clone(); + *max_idle_time = notebook_image.max_idle_time; + + NotebookSpec { + template: NotebookTemplateSpec { + spec: PodSpec { + containers: vec![ContainerSpec { + name, + image: notebook_image.url.clone(), + resources: Some(ResourceRequirements { + requests: BTreeMap::from([ + ("cpu".to_string(), "2".to_string()), + ("memory".to_string(), "4Gi".to_string()), + ]), + limits: BTreeMap::from([ + ("cpu".to_string(), "2".to_string()), + ("memory".to_string(), "4Gi".to_string()), + ]), + }), + image_pull_policy: notebook_image.pull_policy.clone(), + volume_mounts: Some(vec![VolumeMount { + name: volume_name.clone(), + mount_path: notebook_image.volume_mnt_path.clone().unwrap_or_default(), + }]), + command: notebook_image.command.clone(), + args: notebook_image.args.clone(), + workingdir: notebook_image.working_dir.clone(), + env: Some(env), + }], + image_pull_secrets: notebook_image + .secret + .clone() + .map(|secret| vec![ImagePullSecret { name: secret }]), + volumes: Some(vec![VolumeSpec { + name: volume_name.clone(), + persistent_volume_claim: Some(PersistentVolumeClaimSpec { + claim_name: volume_name, + read_only: None, + }), + config_map: None, + }]), + }, + }, + } + } +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct NotebookTemplateSpec { + spec: PodSpec, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct PodSpec { + containers: Vec, + #[serde(rename = "imagePullSecrets")] + image_pull_secrets: Option>, + volumes: Option>, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct ImagePullSecret { + name: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct ContainerSpec { + name: String, + image: String, + resources: Option, + #[serde(rename = "workingDir")] + workingdir: Option, + #[serde(rename = "imagePullPolicy")] + image_pull_policy: String, + #[serde(rename = "volumeMounts")] + volume_mounts: Option>, + command: Option>, + args: Option>, + env: Option>, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +struct ResourceRequirements { + requests: BTreeMap, + limits: BTreeMap, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct EnvVar { + name: String, + value: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct VolumeMount { + name: String, + #[serde(rename = "mountPath")] + mount_path: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct VolumeSpec { + name: String, + #[serde(rename = "persistentVolumeClaim")] + #[serde(skip_serializing_if = "Option::is_none")] + persistent_volume_claim: Option, + #[serde(rename = "configMap")] + #[serde(skip_serializing_if = "Option::is_none")] + config_map: Option, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct ConfigMapSpec { + pub name: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] +pub struct PersistentVolumeClaimSpec { + #[serde(rename = "claimName")] + claim_name: String, + #[serde(rename = "readOnly")] + #[serde(skip_serializing_if = "Option::is_none")] + read_only: Option, +} + +// Define the PVC Spec +pub struct PVCSpec { + pub spec: PersistentVolumeClaim, +} + +impl PVCSpec { + pub fn new(name: String, storage_size: usize) -> Self { + Self { + spec: PersistentVolumeClaim { + metadata: ObjectMeta { + name: Some(name), + ..Default::default() + }, + spec: Some(k8s_openapi::api::core::v1::PersistentVolumeClaimSpec { + access_modes: Some(vec!["ReadWriteOnce".to_string()]), + resources: Some(VolumeResourceRequirements { + requests: Some(BTreeMap::from([( + "storage".to_string(), + Quantity(storage_size.to_string() + "Gi"), + )])), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + }, + } + } +} + +#[cfg(test)] +mod test { + use super::*; + use serde_json::json; + + #[test] + fn test_notebook_spec() { + let name = "notebook".to_string(); + let volume_name = "notebook-volume".to_string(); + let mut start_url = None; + let mut max_idle_time = None; + + let spec = NotebookSpec::new( + name, + "open_ad_workbench", + volume_name, + &mut start_url, + &mut max_idle_time, + vec![], + ); + + // TODO: Remove this test or refactor, so it doesn't break all the time the notebok config + // is updated + let expected = json!({ + "template": { + "spec": { + "containers": [{ + "name": "notebook", + "image": "quay.io/ibmdpdev/openad_workbench_prod:latest", + "resources": { + "requests": { + "cpu": "2", + "memory": "4Gi" + }, + "limits": { + "cpu": "2", + "memory": "4Gi" + } + }, + "workingDir": "/opt/app-root/src", + "imagePullPolicy": "Always", + "volumeMounts": [ + { + "name": "notebook-volume", + "mountPath": "/opt/app-root/src" + } + ], + "command": null, + "args": null, + "env": [ + { + "name": "NOTEBOOK_ARGS", + "value": "--ServerApp.token='' --ServerApp.password='' --ServerApp.notebook_dir='/opt/app-root/src' --ServerApp.quit_button=False --LabApp.default_url='/lab/tree/start_menu.ipynb' --ServerApp.default_url='/lab/tree/start_menu.ipynb' --ServerApp.trust_xheaders=True --ServerApp.base_url='notebook/notebook/notebook'" + }, + { + + "name":"PROXY_URL", + "value": "https://open.accelerate.science/proxy" + } + ], + }], + "imagePullSecrets": [ + { + "name": "ibmdpdev-openad-pull-secret" + } + ], + "volumes": [{ + "name": "notebook-volume", + "persistentVolumeClaim": { + "claimName": "notebook-volume" + } + }] + }, + } + }); + + let actual = serde_json::to_value(&spec).unwrap(); + assert_eq!(actual, expected); + assert_eq!(start_url, Some("lab/tree/start_menu.ipynb".to_string())); + assert_eq!(max_idle_time, Some(86400)); + } + + #[test] + fn test_pvc_spec() { + let name = "notebook-volume".to_string(); + let storage_size = 10; + + let spec = PVCSpec::new(name, storage_size); + + let expected = json!({ + "apiVersion": "v1", + "kind": "PersistentVolumeClaim", + "metadata": { + "name": "notebook-volume" + }, + "spec": { + "accessModes": ["ReadWriteOnce"], + "resources": { + "requests": { + "storage": "10Gi" + } + } + } + }); + + let actual = serde_json::to_value(&spec.spec).unwrap(); + assert_eq!(actual, expected); + } +} diff --git a/src/kube/notebook_lifecycle.rs b/src/kube/notebook/notebook_lifecycle.rs similarity index 99% rename from src/kube/notebook_lifecycle.rs rename to src/kube/notebook/notebook_lifecycle.rs index af0958f5..cab6e401 100644 --- a/src/kube/notebook_lifecycle.rs +++ b/src/kube/notebook/notebook_lifecycle.rs @@ -23,7 +23,7 @@ use crate::{ mongo::{DB, DBCONN, ObjectID}, }, errors::{BridgeError, Result}, - kube::KubeAPI, + kube::{KubeAPI, NOTEBOOK_NAMESPACE}, web::{ notebook_helper::{make_forward_url, make_notebook_name}, utils, @@ -86,6 +86,12 @@ where fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let mut this = self.project(); + // check if we need to shutdown + if this.sigterm.poll(cx).is_ready() { + info!("Shutting down lifecycle"); + return Poll::Ready(()); + } + if !*this.slept { match this.sleep.as_mut().poll(cx) { Poll::Ready(_) => { @@ -97,12 +103,6 @@ where } } - // check if we need to shutdown - if this.sigterm.poll(cx).is_ready() { - info!("Shutting down lifecycle"); - return Poll::Ready(()); - } - let now = OffsetDateTime::now_utc(); if now >= *this.expiration { @@ -234,7 +234,7 @@ pub async fn notebook_lifecycle(client: Client) -> Result<()> { .get() .ok_or(BridgeError::GeneralError("DB connection failed".into()))?; - let pods = KubeAPI::::get_all_pods().await?; + let pods = KubeAPI::::get_all_pods(*NOTEBOOK_NAMESPACE).await?; if pods.is_empty() { info!("No running notebooks found"); return Ok(()); @@ -373,7 +373,6 @@ mod tests { async fn test(client: Client) -> Result<()> { // ping postman echo let resp = client.get("https://postman-echo.com/get").send().await?; - println!("{:?}", resp); assert!(resp.status().is_success()); Ok(()) } diff --git a/src/kube/openwebui/mod.rs b/src/kube/openwebui/mod.rs new file mode 100644 index 00000000..eba0e440 --- /dev/null +++ b/src/kube/openwebui/mod.rs @@ -0,0 +1,10 @@ +#![allow(dead_code)] + +use std::marker::PhantomData; + +pub const OWUI: &str = "owui"; + +// This is a placeholder for openwebui CRD +struct OpenWebUI { + _p: PhantomData<()>, +} diff --git a/src/lib.rs b/src/lib.rs index a868cb6c..172dc056 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,7 +2,7 @@ pub mod auth; pub mod config; pub mod db; pub mod errors; -#[cfg(feature = "notebook")] +#[cfg(feature = "kubernetes")] pub mod kube; pub mod logger; pub mod templating; diff --git a/src/logger/futures.rs b/src/logger/futures.rs new file mode 100644 index 00000000..494880fc --- /dev/null +++ b/src/logger/futures.rs @@ -0,0 +1,75 @@ +use std::{ + future::Future, + pin::Pin, + sync::{Arc, Mutex}, + task::{Context, Poll}, + time::Duration, +}; + +use pin_project::pin_project; +use tokio::{ + sync::{broadcast::error::RecvError, mpsc::Receiver}, + time::{Sleep, sleep}, +}; + +static MAX_CAP: usize = 100; +static WAIT: u64 = 60 * 60; + +#[pin_project] +pub struct FutureRace { + fut: Arc>>, + #[pin] + timer: Sleep, + events: Option>, + #[pin] + term: F, +} + +impl FutureRace { + pub fn new(fut: Arc>>, term: F) -> Self { + // Create a timer that will sleep for 60 minutes + let sleep = sleep(Duration::from_secs(WAIT)); + let events = Vec::with_capacity(MAX_CAP); + Self { + fut, + timer: sleep, + events: Some(events), + term, + } + } +} + +impl Future for FutureRace +where + F: Future>, +{ + type Output = Option>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let mut this = self.project(); + + if this.timer.poll(cx).is_ready() { + return Poll::Ready(this.events.take()); + } + + // the channel is buffered and multiple events can be received, so we should drain it + loop { + if this.term.as_mut().poll(cx).is_ready() { + return Poll::Ready(this.events.take()); + } + + match this.fut.lock().unwrap().poll_recv(cx) { + Poll::Ready(Some(event)) => { + if let Some(events) = this.events.as_mut() { + events.push(event); + if events.len() >= MAX_CAP { + return Poll::Ready(this.events.take()); + } + } + } + Poll::Ready(None) => return Poll::Ready(this.events.take()), + Poll::Pending => return Poll::Pending, + } + } + } +} diff --git a/src/logger/mod.rs b/src/logger/mod.rs index ba1dff6c..cb14fb68 100644 --- a/src/logger/mod.rs +++ b/src/logger/mod.rs @@ -1,12 +1,19 @@ use reqwest::Client; +use tokio::sync::broadcast::Sender; use tracing_subscriber::{filter::LevelFilter, prelude::*}; +#[cfg(feature = "observe")] +mod futures; #[cfg(feature = "observe")] mod observability; #[cfg(feature = "observe")] pub use observability::MESSAGE_DELIMITER; +#[cfg(feature = "observe")] +pub use observability::PERSIST_META; +#[cfg(feature = "observe")] +pub use observability::PERSIST_TIME; -pub fn start_logger(level: LevelFilter, _client: Client) { +pub fn start_logger(level: LevelFilter, _client: Client, _tx: Sender<()>) { // let file = std::fs::File::create("./log").unwrap(); let ts = tracing_subscriber::registry().with( @@ -24,9 +31,16 @@ pub fn start_logger(level: LevelFilter, _client: Client) { use crate::config::CONFIG; if let Some((ref api_key, ref endpoint)) = CONFIG.observability_cred { + use crate::db::mongo::DBCONN; + let writer = observability::Observe::new(api_key, endpoint, _client) .expect("Failed to create observability for logger"); - ts.with(writer.wrap_layer(level)) + let observe_layer = observability::ObserveEvents::new( + DBCONN.get().expect("DB connection not initialized"), + _tx, + ); + + ts.with(writer.wrap_layer(level)).with(observe_layer) } else { panic!("Observability credentials are not set in the configuration") } diff --git a/src/logger/observability.rs b/src/logger/observability.rs index 05caecb7..15237cae 100644 --- a/src/logger/observability.rs +++ b/src/logger/observability.rs @@ -1,12 +1,18 @@ -use std::io::{self, Write}; +use std::{ + io::{self, Write}, + sync::{Arc, Mutex}, +}; use reqwest::{Client, header::CONTENT_TYPE}; use serde_json::json; use tokio::{ - sync::mpsc::{Sender, channel}, + sync::{ + broadcast::Sender as BSender, + mpsc::{Sender, channel}, + }, task::JoinHandle, }; -use tracing::{error, level_filters::LevelFilter, warn}; +use tracing::{Subscriber, error, field::Visit, level_filters::LevelFilter, warn}; use tracing_subscriber::{ Layer, Registry, filter, fmt::{self, MakeWriter, format}, @@ -14,15 +20,32 @@ use tracing_subscriber::{ }; use url::Url; -use crate::errors::{BridgeError, Result}; +use crate::{ + db::{ + Database, + models::{OBSERVE, ObserveEventEntry}, + mongo::{DB, helper::i64_to_bson_datatime}, + }, + errors::{BridgeError, Result}, +}; + +use super::futures::FutureRace; pub const MESSAGE_DELIMITER: &str = "*~*~*"; +pub const PERSIST_META: &str = "persist_to_db"; +// 6 months +pub const PERSIST_TIME: i64 = 15778463; pub struct Observe { sender: Sender, handler: Option>, } +pub struct ObserveEvents { + sender: Sender, + handler: Option>, +} + type LayerAlias = filter::Filtered< fmt::Layer< layer::Layered< @@ -48,7 +71,15 @@ type LayerAlias = filter::Filtered< >, >; -const CHANNEL_SIZE: usize = 100; +const CHANNEL_SIZE: usize = 150; + +struct ObserveEventTrace { + sub: Option, + // group: Option, + property: Option, + request_date: Option, + expire_soon_after: Option, +} impl Observe { /// Warning! Ensure you don't instrustment this with sensitive data, as it will be sent to the @@ -62,7 +93,6 @@ impl Observe { let endpoint = endpoint.as_str(); let api_key = api_key; while let Some(msg) = recv.recv().await { - println!("Sending observability message: {}", msg); let msg = json!( { "text": msg @@ -116,6 +146,61 @@ impl Observe { } } +impl ObserveEvents { + pub fn new(db: &'static DB, tx: BSender<()>) -> Self { + let (sender, recv) = channel(CHANNEL_SIZE); + + let handler = tokio::spawn(async move { + let recv = Arc::new(Mutex::new(recv)); + let mut term_outer = tx.subscribe(); + loop { + let mut term_inner = tx.subscribe(); + let get_events = FutureRace::new(recv.clone(), term_inner.recv()); + if let Some(events) = get_events.await { + if events.is_empty() { + continue; + } + if let Err(e) = db.insert_many(events, OBSERVE).await { + error!("Failed to insert observability events: {}", e); + } + // shutdown stop + if term_outer.try_recv().is_ok() { + break; + } + } + } + }); + + Self { + sender, + handler: Some(handler), + } + } + + pub fn send_message(&self, event: ObserveEventEntry) { + let message = format!("{:?}", &event); + if let Err(e) = self.sender.try_send(event) { + match e { + tokio::sync::mpsc::error::TrySendError::Full(_) => warn!( + "Observability channel is full, dropping message: {:?}", + message + ), + tokio::sync::mpsc::error::TrySendError::Closed(_) => error!( + "Observability channel is closed, cannot send message: {:?}", + message + ), + } + } + } + + #[allow(dead_code)] + pub async fn close(mut self) -> Result<()> { + let handler = self.handler.take(); + drop(self); + Ok(handler.unwrap().await?) + } +} + impl Write for &Observe { fn write(&mut self, buf: &[u8]) -> io::Result { let message = String::from_utf8(buf.to_vec()).unwrap_or_default(); @@ -141,13 +226,68 @@ impl<'a> MakeWriter<'a> for Observe { } } +impl Visit for ObserveEventTrace { + fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { + if field.name() == "request_date" { + self.request_date = Some(value); + } else if field.name() == "expire_soon_after" { + self.expire_soon_after = Some(value); + } else { + unreachable!("unimplemented field: {}", field.name()); + } + } + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + match field.name() { + "sub" => self.sub = Some(value.to_string()), + // "group" => self.group = Some(value.to_string()), + "property" => self.property = Some(value.to_string()), + _ => unreachable!("unimplmented field: {}", field.name()), + } + } + // NOOPS + fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {} +} + +impl Layer for ObserveEvents +where + S: Subscriber, +{ + fn on_event(&self, event: &tracing::Event<'_>, _ctx: layer::Context<'_, S>) { + if event.metadata().target() == PERSIST_META { + let mut trace = ObserveEventTrace { + sub: None, + // group: None, + property: None, + request_date: None, + expire_soon_after: None, + }; + event.record(&mut trace); + + if let (Some(request_date), Some(expire_soon_after)) = + (trace.request_date, trace.expire_soon_after) + { + let entry = ObserveEventEntry { + sub: trace.sub.take().unwrap_or_default(), + // group: trace.group.take().unwrap_or_default(), + property: trace.property.take().unwrap_or_default(), + request_date: time::OffsetDateTime::from_unix_timestamp(request_date) + .unwrap_or(time::OffsetDateTime::now_utc()), + expire_soon_after: i64_to_bson_datatime(expire_soon_after), + }; + self.send_message(entry); + } + } + } +} + #[cfg(test)] mod tests { use super::*; #[tokio::test] async fn test_observability() { - let observe = Observe::new("api-key", "https://api.slack.com/api", Client::new()).unwrap(); + let observe = + Observe::new("api-key", "https://api.slack.com/api", Client::new()).unwrap(); let mut writer = &observe; let _ = writer .write(b"Nothing*~*~*Hello there from open bridge") diff --git a/src/web/bridge_middleware/authenicate.rs b/src/web/bridge_middleware/authenicate.rs index 6d6afb0b..3f6901a0 100644 --- a/src/web/bridge_middleware/authenicate.rs +++ b/src/web/bridge_middleware/authenicate.rs @@ -3,7 +3,11 @@ use actix_web_httpauth::extractors::{ AuthenticationError, bearer::{self, BearerAuth}, }; +#[cfg(feature = "observe")] +use tracing::info; +#[cfg(feature = "observe")] +use crate::logger::{PERSIST_META, PERSIST_TIME}; use crate::{auth::jwt::validate_token, config::CONFIG, web::route::proxy::INFERENCE_HEADER}; pub async fn validator( @@ -25,6 +29,14 @@ pub async fn validator( // TODO: handle unwrap_or better let inference = r.to_str().unwrap_or(""); if claims.scp.contains(&inference.to_string()) { + + #[cfg(feature = "observe")] + { + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + info!(target: PERSIST_META, + sub=claims.get_sub(), property=inference, request_date=now, expire_soon_after=now+PERSIST_TIME); + } + return Ok(req); } } diff --git a/src/web/bridge_middleware/cookie_check.rs b/src/web/bridge_middleware/cookie_check.rs index 58ef2997..d2e791a9 100644 --- a/src/web/bridge_middleware/cookie_check.rs +++ b/src/web/bridge_middleware/cookie_check.rs @@ -71,6 +71,19 @@ where } } + #[cfg(feature = "openwebui")] + { + use crate::{auth::OWUI_COOKIE_NAME, db::models::OWUICookie}; + + if let Some(owui_cookie) = req.cookie(OWUI_COOKIE_NAME) { + if let Ok(oc) = + serde_json::from_str::(owui_cookie.value()) + { + req.extensions_mut().insert(oc); + } + } + } + req.extensions_mut().insert(gcs.clone()); let service = self.service.clone(); diff --git a/src/web/bridge_middleware/mod.rs b/src/web/bridge_middleware/mod.rs index 295b5a63..4855aa04 100644 --- a/src/web/bridge_middleware/mod.rs +++ b/src/web/bridge_middleware/mod.rs @@ -4,7 +4,10 @@ mod errors; mod htmx; mod https; mod maintenance; +#[cfg(feature = "notebook")] mod notebook_cookie_check; +#[cfg(feature = "openwebui")] +mod openwebui_cookie_check; mod resource_cookie_check; mod security_cache_header; @@ -15,6 +18,8 @@ pub use htmx::{HTMX_ERROR_RES, Htmx}; pub use maintenance::{MAINTENANCE_WINDOWS, Maintainence}; #[cfg(feature = "notebook")] pub use notebook_cookie_check::NotebookCookieCheck; +#[cfg(feature = "openwebui")] +pub use openwebui_cookie_check::OWUICookieCheck; pub use resource_cookie_check::ResourceCookieCheck; pub use security_cache_header::SecurityCacheHeader; diff --git a/src/web/bridge_middleware/openwebui_cookie_check.rs b/src/web/bridge_middleware/openwebui_cookie_check.rs new file mode 100644 index 00000000..4208c201 --- /dev/null +++ b/src/web/bridge_middleware/openwebui_cookie_check.rs @@ -0,0 +1,79 @@ +use std::future::{Ready, ready}; + +use actix_web::{ + Error, HttpMessage, HttpResponse, + body::EitherBody, + dev::{Service, ServiceRequest, ServiceResponse, Transform, forward_ready}, +}; +use futures::{FutureExt, TryFutureExt, future::LocalBoxFuture}; +use tracing::warn; + +use crate::{auth::OWUI_COOKIE_NAME, db::models::OWUICookie}; + +pub struct OWUICookieCheck; + +impl Transform for OWUICookieCheck +where + S: Service, Error = Error>, + S::Future: 'static, + B: 'static, +{ + type Response = ServiceResponse>; + type Error = Error; + type InitError = (); + type Transform = OWUICookieCheckMW; + type Future = Ready>; + + fn new_transform(&self, service: S) -> Self::Future { + ready(Ok(OWUICookieCheckMW { service })) + } +} + +pub struct OWUICookieCheckMW { + service: S, +} + +impl Service for OWUICookieCheckMW +where + S: Service, Error = Error>, + S::Future: 'static, + B: 'static, +{ + type Response = ServiceResponse>; + type Error = Error; + type Future = LocalBoxFuture<'static, Result>; + + forward_ready!(service); + + fn call(&self, req: ServiceRequest) -> Self::Future { + match req.cookie(OWUI_COOKIE_NAME).map(|c| c.value().to_string()) { + Some(v) => { + let bridge_cookie_result = serde_json::from_str::(&v); + match bridge_cookie_result { + Ok(gcs) => { + req.extensions_mut().insert(gcs); + self.service + .call(req) + .map_ok(ServiceResponse::map_into_left_body) + .boxed_local() + } + Err(e) => { + warn!("OWUI cookie deserialization error: {:?}", e); + let res = HttpResponse::InternalServerError() + .finish() + .map_into_right_body(); + Box::pin(async { Ok(req.into_response(res)) }) + } + } + } + None => { + // warn!( + // "OWUI cookie not found from ip {:?}", + // req.connection_info().realip_remote_addr() + // ); + let res = HttpResponse::Forbidden().finish().map_into_right_body(); + Box::pin(async { Ok(req.into_response(res)) }) + } + } + } +} diff --git a/src/web/helper.rs b/src/web/helper.rs index adf1e99c..1a7bbfd6 100644 --- a/src/web/helper.rs +++ b/src/web/helper.rs @@ -1,6 +1,8 @@ use actix_web::web; use base64::{Engine, prelude::BASE64_STANDARD}; use mongodb::bson::{Bson, to_bson}; +#[cfg(feature = "observe")] +use mongodb::bson; use rand::{Rng, rng}; use serde::Deserialize; use tera::Context; @@ -217,6 +219,7 @@ pub mod forwarding { use crate::errors::{BridgeError, Result}; // No inline needed... generic are inherently inlined + #[allow(clippy::too_many_arguments)] pub async fn forward( req: HttpRequest, mut payload: web::Payload, @@ -225,6 +228,7 @@ pub mod forwarding { client: web::Data, new_url: T, updated_cookie: Option>, + inference: bool, ) -> Result where T: AsRef + Send + Sync, @@ -273,9 +277,13 @@ pub mod forwarding { headers.insert("X-Forwarded-For", ip); } } + headers.insert( + "X-Forwarded-Proto", + ReqwestHeaderValue::from_static("https"), + ); for (header_name, header_value) in req.headers().iter() { - if header_name == "authorization" || header_name == "inference-service" { + if inference && (header_name == "authorization" || header_name == "inference-service") { continue; } headers.insert( @@ -286,13 +294,10 @@ pub mod forwarding { let forwarded_req = forwarded_req.headers(headers); - let res = log_with_level!( - forwarded_req - .send() - .await - .map_err(|e| { BridgeError::GeneralError(e.to_string()) }), - error - )?; + let res = forwarded_req.send().await.map_err(|e| { + error!("{:?}", e); + BridgeError::GeneralError(e.to_string()) + })?; let status = res.status().as_u16(); let status = @@ -308,11 +313,11 @@ pub mod forwarding { // https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Connection#Directives // Also removing "content-length" since we are streaming the response, and content-length // is not necessarily needed. - for (header_name, header_value) in res - .headers() - .iter() - .filter(|(h, _)| *h != "connection" && *h != "keep-alive" && *h != "content-length") - { + for (header_name, header_value) in res.headers().iter().filter(|(h, _)| { + h.as_str().to_lowercase() != "connection" + && h.as_str().to_lowercase() != "keep-alive" + && h.as_str().to_lowercase() != "content-length" + }) { // Again copy over seem incredibly inefficient. It sure is, but like before, we do this // because actix-web and reqwest use different versions of http. Once Actix-web // updates their http version, we can remove this. @@ -343,7 +348,7 @@ pub mod utils { mongo::ObjectID, }, errors::Result, - kube::{KubeAPI, Notebook}, + kube::{KubeAPI, NOTEBOOK_NAMESPACE, Notebook}, web::{helper::bson, notebook_helper}, }; @@ -370,10 +375,13 @@ pub mod utils { { let name = notebook_helper::make_notebook_name(subject); let pvc_name = notebook_helper::make_notebook_volume_name(subject); - log_with_level!(KubeAPI::::delete(&name).await, error)?; + log_with_level!( + KubeAPI::::delete(&name, *NOTEBOOK_NAMESPACE).await, + error + )?; if !persist_pvc { log_with_level!( - KubeAPI::::delete(&pvc_name).await, + KubeAPI::::delete(&pvc_name, *NOTEBOOK_NAMESPACE).await, error )?; } @@ -411,7 +419,10 @@ pub mod ws { use tokio_tungstenite::tungstenite::{ self, handshake::client::Request, - protocol::frame::coding::{Data, OpCode}, + protocol::{ + WebSocketConfig, + frame::coding::{Data, OpCode}, + }, }; use tracing::{error, warn}; @@ -439,7 +450,16 @@ pub mod ws { } let request = request.body(()).unwrap(); - let (stream, res) = tokio_tungstenite::connect_async(request).await.unwrap(); + let (stream, res) = tokio_tungstenite::connect_async_with_config( + request, + Some(WebSocketConfig::default().accept_unmasked_frames(true)), + false, + ) + .await + .map_err(|e| { + error!("WebSocket connection error: {:?}", e); + BridgeError::GeneralError(e.to_string()) + })?; if !res.status().eq(&StatusCode::SWITCHING_PROTOCOLS) { return Err(BridgeError::GeneralError( "Failed to establish websocket connection".to_string(), diff --git a/src/web/mod.rs b/src/web/mod.rs index 6d259bb7..52a72889 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -1,22 +1,20 @@ use std::{io::Result, process::exit, time::Duration}; -#[cfg(all(feature = "notebook", feature = "lifecycle"))] -use std::pin::pin; - +#[cfg(feature = "openwebui")] +use actix_web::guard; use actix_web::{ App, HttpServer, middleware::{self}, web::{self, Data}, }; use tera::Context; -use tracing::{error, level_filters::LevelFilter, warn}; +use tokio::sync::broadcast::channel; +use tracing::level_filters::LevelFilter; #[cfg(feature = "notebook")] use crate::kube::{self}; #[cfg(all(feature = "notebook", feature = "lifecycle"))] use crate::kube::{LifecycleStream, Medium, notebook_lifecycle}; -#[cfg(all(feature = "notebook", feature = "lifecycle"))] -use futures::future::select; use crate::{ auth::openid, @@ -71,13 +69,6 @@ pub async fn start_server(with_tls: bool) -> Result<()> { .build() .expect("Cannot create reqwest client"); - // Logger and this is not configurable by the caller - if cfg!(debug_assertions) { - logger::start_logger(LevelFilter::DEBUG, client.clone()); - } else { - logger::start_logger(LevelFilter::INFO, client.clone()); - } - rustls::crypto::ring::default_provider() .install_default() .expect("Cannot install default provider with ring"); @@ -85,49 +76,43 @@ pub async fn start_server(with_tls: bool) -> Result<()> { // Singletons openid::init_once().await; if let Err(e) = DB::init_once(&DBNAME).await { - error!("{e}"); + eprintln!("{e}"); exit(1); } if let Err(e) = CacheDB::init_once().await { // we don't want to make caching a hard requirement for now - warn!("{e}: continuing without caching"); + eprintln!("{e}: continuing without caching"); } let db = match DBCONN.get() { Some(db) => db, None => { - error!("DB Connection not found... Is the DB running?"); + eprintln!("DB Connection not found... Is the DB running?"); exit(1); } }; #[cfg(feature = "notebook")] kube::init_once().await; - // Maintainence window impl + #[allow(unused_mut)] + #[allow(unused_variables)] + let (sender, mut recv) = channel::<()>(1); + let tx = sender.clone(); + + // Logger and this is not configurable by the caller + if cfg!(debug_assertions) { + logger::start_logger(LevelFilter::DEBUG, client.clone(), tx); + } else { + logger::start_logger(LevelFilter::INFO, client.clone(), tx); + } + + // Launch maintainence window watcher if cache is available let _ = maintenance_watch(); // Lifecycle with "advisory lock" #[cfg(all(feature = "notebook", feature = "lifecycle"))] let handle = tokio::spawn(async move { let stream = LifecycleStream::new(notebook_lifecycle); - Medium::new( - LIFECYCLE_TIME, - SIGTERM_FREQ, - db, - stream, - select( - pin!( - tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) - .expect("Cannot establish SIGTERM signal") - .recv() - ), - pin!( - tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) - .expect("Cannot establish SIGINT signal") - .recv() - ), - ), - ) - .await; + Medium::new(LIFECYCLE_TIME, SIGTERM_FREQ, db, stream, recv.recv()).await; }); let server = HttpServer::new(move || { @@ -152,16 +137,36 @@ pub async fn start_server(with_tls: bool) -> Result<()> { .app_data(client_data) .app_data(db) .app_data(cache) - .wrap(bridge_middleware::custom_code_handle(tera_data, context)) .wrap(middleware::NormalizePath::trim()) .wrap(middleware::Compress::default()) - .wrap(bridge_middleware::Maintainence) - .service(actix_files::Files::new("/static", "static")); + .wrap(bridge_middleware::Maintainence); + + #[cfg(feature = "openwebui")] + let app = { + use self::bridge_middleware::OWUICookieCheck; + app.service( + web::scope("") + .guard(guard::Host(&CONFIG.openweb_url)) + .wrap(OWUICookieCheck) + .configure(route::openwebui::config_openwebui), + ) + .service( + web::scope("") + .guard(guard::Host(&CONFIG.moleviewer_url)) + .wrap(OWUICookieCheck) + .configure(route::openwebui::config_moleviewer), + ) + }; + + let app = app.service(actix_files::Files::new("/static", "static")); + #[cfg(feature = "notebook")] let app = app.configure(route::notebook::config_notebook); + app.service({ let scope = web::scope("") .wrap(bridge_middleware::SecurityCacheHeader) + .wrap(bridge_middleware::custom_code_handle(tera_data, context)) .configure(route::auth::config_auth) .configure(route::health::config_status) .configure(route::proxy::config_proxy) @@ -203,6 +208,9 @@ pub async fn start_server(with_tls: bool) -> Result<()> { server.bind(("0.0.0.0", 8080))?.run().await?; } + // shutdown signal + sender.send(()).unwrap(); + // If the lock was acquired, release it #[cfg(all(feature = "notebook", feature = "lifecycle"))] handle.await?; diff --git a/src/web/route/auth/mod.rs b/src/web/route/auth/mod.rs index b0fad8f1..fa475d56 100644 --- a/src/web/route/auth/mod.rs +++ b/src/web/route/auth/mod.rs @@ -221,7 +221,7 @@ async fn code_to_response( }; let content = serde_json::to_string(&bridge_cookie_json).map_err(|e| { - BridgeError::GeneralError(format!("Could not serialize bridge cookie: {}", e)) + BridgeError::GeneralError(format!("Could not serialize bridge cookie: {e}")) })?; // create cookie for all routes for this user diff --git a/src/web/route/auth/oauth.rs b/src/web/route/auth/oauth.rs index 86c3319b..14729ea3 100644 --- a/src/web/route/auth/oauth.rs +++ b/src/web/route/auth/oauth.rs @@ -103,7 +103,7 @@ pub async fn register_app( pl: web::Payload, db: Data<&DB>, ) -> Result { - // check the content type is jpon + // check the content type is json if !req.content_type().eq(ContentType::json().as_ref()) { return Ok(HttpResponse::UnsupportedMediaType().finish()); } diff --git a/src/web/route/health/inference_services.rs b/src/web/route/health/inference_services.rs index c92b469a..1e291908 100644 --- a/src/web/route/health/inference_services.rs +++ b/src/web/route/health/inference_services.rs @@ -96,10 +96,9 @@ impl ListBuilder<'_> { }; self.inner_body.push_str(&format!( - r##"
{}
-
Service is currently {}.
-
{} ms
"##, - name, status, state, elapsed + r##"
{name}
+
Service is currently {status}.
+
{elapsed} ms
"## )); } diff --git a/src/web/route/mcp/mod.rs b/src/web/route/mcp/mod.rs index 72bfaccc..e2afa793 100644 --- a/src/web/route/mcp/mod.rs +++ b/src/web/route/mcp/mod.rs @@ -1,7 +1,11 @@ use actix_web::{HttpRequest, HttpResponse, dev::PeerAddr, http::Method, web}; use actix_web_httpauth::extractors::bearer::BearerAuth; use tracing::{instrument, warn}; +#[cfg(feature = "observe")] +use tracing::info; +#[cfg(feature = "observe")] +use crate::logger::{PERSIST_META, PERSIST_TIME}; use crate::{ auth::jwt::validate_token, config::CONFIG, @@ -10,6 +14,7 @@ use crate::{ }; const MCP_PREFIX: &str = "/mcp/"; +const MCP_SUFFIX: &str = "-s"; #[instrument(skip(payload))] async fn forward( @@ -41,6 +46,13 @@ async fn forward( "User does not have access to {mcp}" ))); } + + #[cfg(feature = "observe")] + { + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + info!(target: PERSIST_META, + sub=claims.get_sub(), property=mcp, request_date=now, expire_soon_after=now+PERSIST_TIME); + } } else { return Err(BridgeError::Unauthorized( "JWT token is invalid".to_string(), @@ -48,10 +60,16 @@ async fn forward( } let mut new_url = helper::log_with_level!(CATALOG.get_service(mcp), error)?; - new_url.set_path(path); + // fastmcp temp(?) fix + if mcp.ends_with(MCP_SUFFIX) { + new_url.set_path(&format!("{path}/")); + } else { + new_url.set_path(path); + } new_url.set_query(req.uri().query()); - helper::forwarding::forward(req, payload, method, peer_addr, client, new_url, None).await + helper::forwarding::forward(req, payload, method, peer_addr, client, new_url, None, true) + .await } else { warn!("MCP service not found in url request"); Err(BridgeError::MCPParseIssue) diff --git a/src/web/route/mod.rs b/src/web/route/mod.rs index 31d3b931..0aa12bdc 100644 --- a/src/web/route/mod.rs +++ b/src/web/route/mod.rs @@ -18,6 +18,8 @@ pub mod health; pub mod mcp; #[cfg(feature = "notebook")] pub mod notebook; +#[cfg(feature = "openwebui")] +pub mod openwebui; pub mod portal; pub mod proxy; pub mod resource; diff --git a/src/web/route/notebook/mod.rs b/src/web/route/notebook/mod.rs index d40f2dfb..e6fae9d7 100644 --- a/src/web/route/notebook/mod.rs +++ b/src/web/route/notebook/mod.rs @@ -34,7 +34,7 @@ use crate::{ mongo::{DB, ObjectID}, }, errors::{BridgeError, Result}, - kube::{KubeAPI, NAMESPACE, Notebook, NotebookSpec, PVCSpec}, + kube::{KubeAPI, NOTEBOOK_NAMESPACE, Notebook, NotebookSpec, PVCSpec}, web::{ bridge_middleware::{CookieCheck, Htmx, NotebookCookieCheck}, helper::{self, bson}, @@ -46,6 +46,7 @@ const NOTEBOOK_PORT: &str = "8888"; const NOTEBOOK_TOKEN_LIFETIME: usize = const { 60 * 60 * 24 * 30 }; const PVC_DELETE_ATTEMPT: u8 = 9; +// TODO: name is not needed... look into removing them if possible #[get("{name}/api/events/subscribe")] async fn notebook_ws_subscribe( req: HttpRequest, @@ -101,10 +102,7 @@ async fn notebook_ws_session( let kernel_id = kernel.into_inner().1; let session_id = session_id.session_id.clone(); - let path = format!( - "api/kernels/{}/channels?session_id={}", - kernel_id, session_id - ); + let path = format!("api/kernels/{kernel_id}/channels?session_id={session_id}"); let url = notebook_helper::make_forward_url( ¬ebook_cookie.ip, ¬ebook_helper::make_notebook_name(¬ebook_cookie.subject), @@ -190,7 +188,7 @@ async fn notebook_create( if user.notebook.is_some() { warn!("Notebook already exists for user {}", bridge_cookie.subject); - return Err(BridgeError::NotebookExistsError( + return Err(BridgeError::CRDExistsError( "Notebook already exists".to_string(), )); }; @@ -203,7 +201,7 @@ async fn notebook_create( )? .is_some() { - info!("Namespace {} has been created", *NAMESPACE) + info!("Namespace {} has been created", *NOTEBOOK_NAMESPACE) } // User is allowed to create a notebook, but notebook does not exist... so create one @@ -212,14 +210,19 @@ async fn notebook_create( if req.query_string().contains("clear") { helper::log_with_level!( - KubeAPI::::delete(&pvc_name).await, + KubeAPI::::delete(&pvc_name, *NOTEBOOK_NAMESPACE).await, error )?; // PVC takes time to delete... loop and check it is gone loop { let mut loop_cnt = 0; - if KubeAPI::::check_pvc_exists(&pvc_name).await? { + if KubeAPI::::check_pvc_exists( + &pvc_name, + *NOTEBOOK_NAMESPACE, + ) + .await? + { tokio::time::sleep(Duration::from_secs(2)).await; loop_cnt += 1; } else { @@ -240,9 +243,9 @@ async fn notebook_create( } let pvc = PVCSpec::new(pvc_name.clone(), 1); - if let Err(e) = KubeAPI::new(pvc.spec).create().await { + if let Err(e) = KubeAPI::new(pvc.spec).create(*NOTEBOOK_NAMESPACE).await { match e { - BridgeError::NotebookExistsError(_) => info!( + BridgeError::CRDExistsError(_) => info!( "PVC {} already exists, most likely persisted from last session, reusing", pvc_name ), @@ -264,7 +267,10 @@ async fn notebook_create( vec![("PROXY_KEY".to_string(), notebook_token)], ), ); - helper::log_with_level!(KubeAPI::new(notebook).create().await, error)?; + helper::log_with_level!( + KubeAPI::new(notebook).create(*NOTEBOOK_NAMESPACE).await, + error + )?; let current_time = time::OffsetDateTime::now_utc(); db.update( @@ -296,7 +302,7 @@ async fn notebook_create( } let bridge_cookie_json = serde_json::to_string(&bridge_cookie).map_err(|e| { - BridgeError::GeneralError(format!("Could not serialize bridge cookie: {}", e)) + BridgeError::GeneralError(format!("Could not serialize bridge cookie: {e}")) })?; let notebook_cookie = NotebookCookie { subject: bridge_cookie.subject, @@ -308,10 +314,10 @@ async fn notebook_create( start_url: start_up_url, }; let notebook_json = serde_json::to_string(¬ebook_cookie).map_err(|e| { - BridgeError::GeneralError(format!("Could not serialize notebook cookie: {}", e)) + BridgeError::GeneralError(format!("Could not serialize notebook cookie: {e}")) })?; let notebook_status_json = serde_json::to_string(¬ebook_status_cookie).map_err(|e| { - BridgeError::GeneralError(format!("Could not serialize notebook status cookie: {}", e)) + BridgeError::GeneralError(format!("Could not serialize notebook status cookie: {e}")) })?; // Create notebook cookies @@ -430,8 +436,6 @@ async fn notebook_delete( } } - println!("Context: {:?}", ctx); - let content = helper::log_with_level!(data.render("components/notebook/start.html", &ctx), error)?; @@ -464,12 +468,14 @@ async fn notebook_status( // check status on k8s let ready = KubeAPI::::check_pod_running( &(notebook_helper::make_notebook_name(&bridge_cookie.subject) + "-0"), + *NOTEBOOK_NAMESPACE, ) .await?; if ready { let ip = KubeAPI::::get_pod_ip( &(notebook_helper::make_notebook_name(&bridge_cookie.subject) + "-0"), + *NOTEBOOK_NAMESPACE, ) .await?; @@ -503,12 +509,11 @@ async fn notebook_status( let notebook_status_json = serde_json::to_string(¬ebook_status_cookie).map_err(|er| { BridgeError::GeneralError(format!( - "Could not serialize notebook status cookie: {}", - er + "Could not serialize notebook status cookie: {er}" )) })?; let notebook_cookie_json = serde_json::to_string(¬ebook_cookie).map_err(|er| { - BridgeError::GeneralError(format!("Could not serialize notebook cookie: {}", er)) + BridgeError::GeneralError(format!("Could not serialize notebook cookie: {er}")) })?; // TODO: We are leveraing cookies to avoid DB calls... but look into not doing this anymore @@ -666,22 +671,25 @@ async fn notebook_forward( new_url.set_path(path); new_url.set_query(req.uri().query()); - helper::forwarding::forward(req, payload, method, peer_addr, client, new_url, None).await + helper::forwarding::forward( + req, payload, method, peer_addr, client, new_url, None, false, + ) + .await } pub mod notebook_helper { use crate::{ db::models::{BridgeCookie, NotebookInfo, NotebookStatusCookie, User, UserNotebook}, - kube::NAMESPACE, + kube::NOTEBOOK_NAMESPACE, web::route::notebook::NOTEBOOK_PORT, }; pub(crate) fn make_notebook_name(subject: &str) -> String { - format!("{}-notebook", subject) + format!("{subject}-notebook") } pub(crate) fn make_notebook_volume_name(subject: &str) -> String { - format!("{}-notebook-volume-pvc", subject) + format!("{subject}-notebook-volume-pvc") } pub(crate) fn make_forward_url( @@ -694,11 +702,11 @@ pub mod notebook_helper { return match path { Some(p) => format!( "{}://localhost:{}/notebook/{}/{}/{}", - protocol, NOTEBOOK_PORT, *NAMESPACE, name, p + protocol, NOTEBOOK_PORT, *NOTEBOOK_NAMESPACE, name, p ), None => format!( "{}://localhost:{}/notebook/{}/{}", - protocol, NOTEBOOK_PORT, *NAMESPACE, name + protocol, NOTEBOOK_PORT, *NOTEBOOK_NAMESPACE, name ), }; } @@ -706,19 +714,19 @@ pub mod notebook_helper { // TODO: This is super cumbersome... FIX IT FIX IT! Some(p) => format!( "{}://{}:{}/notebook/{}/{}/{}", - protocol, ip, NOTEBOOK_PORT, *NAMESPACE, name, p + protocol, ip, NOTEBOOK_PORT, *NOTEBOOK_NAMESPACE, name, p ), None => format!( "{}://{}:{}/notebook/{}/{}", - protocol, ip, NOTEBOOK_PORT, *NAMESPACE, name + protocol, ip, NOTEBOOK_PORT, *NOTEBOOK_NAMESPACE, name ), } } pub(super) fn make_path(name: &str, path: Option<&str>) -> String { match path { - Some(p) => format!("/notebook/{}/{}/{}", *NAMESPACE, name, p), - None => format!("/notebook/{}/{}", *NAMESPACE, name), + Some(p) => format!("/notebook/{}/{}/{}", *NOTEBOOK_NAMESPACE, name, p), + None => format!("/notebook/{}/{}", *NOTEBOOK_NAMESPACE, name), } } @@ -774,7 +782,7 @@ pub mod notebook_helper { pub fn config_notebook(cfg: &mut web::ServiceConfig) { cfg.service( - web::scope(&("/notebook/".to_string() + *NAMESPACE)) + web::scope(&("/notebook/".to_string() + *NOTEBOOK_NAMESPACE)) .wrap(NotebookCookieCheck) .service(notebook_ws_subscribe) .service(notebook_ws_session) @@ -793,8 +801,25 @@ pub fn config_notebook(cfg: &mut web::ServiceConfig) { #[cfg(test)] mod test { + use crate::config::init_once; + #[tokio::test] async fn test_notebook_forward() { // assert!(true); } + + #[test] + fn test_make_forward_url() { + init_once(); + let url = + super::notebook_helper::make_forward_url("0.0.0.0", "test-notebook", "http", None); + assert_eq!( + url, + format!( + "http://localhost:{}/notebook/{}/test-notebook", + super::NOTEBOOK_PORT, + *super::NOTEBOOK_NAMESPACE + ) + ); + } } diff --git a/src/web/route/openwebui/mod.rs b/src/web/route/openwebui/mod.rs new file mode 100644 index 00000000..e65d45fa --- /dev/null +++ b/src/web/route/openwebui/mod.rs @@ -0,0 +1,154 @@ +use std::{collections::HashSet, str::FromStr, sync::LazyLock}; + +use actix_web::{ + HttpRequest, HttpResponse, + dev::PeerAddr, + get, + http::Method, + web::{self, ReqData}, +}; +use tracing::instrument; +use url::Url; + +use crate::{ + config::CONFIG, + db::models::OWUICookie, + errors::{BridgeError, Result}, + web::helper, +}; + +const OWUI_PORT: &str = "8080"; +const MOLE_VIEW_PORT: &str = "8024"; + +pub static OWUI_NAMESPACE: LazyLock<&str> = LazyLock::new(|| &CONFIG.owui_namespace); +static WHITELIST_ENDPOINTS: LazyLock> = LazyLock::new(|| { + HashSet::from([ + "/api/v1/auths", + "/api/v1/channels", + "/api/v1/folders", + "/api/v1/tools", + "/api/v1/chats", + "/api/v1/knowledge", + "/api/v1/models", + "/api/v1/groups", + "/api/v1/prompts", + "/api/v1/users", + "/api/v1/functions", + "/api/v1/files", + ]) +}); + +#[get("ws/socket.io")] +async fn openwebui_ws( + req: HttpRequest, + pl: web::Payload, + owiu_cookie: Option>, +) -> Result { + let owui_cookie = match owiu_cookie { + Some(cookie) => cookie.into_inner(), + None => { + return Err(BridgeError::Unauthorized( + "OWUI cookie not found".to_string(), + )); + } + }; + + let mut url = Url::from_str(&make_forward_url("ws", &owui_cookie.subject, false))?; + url.set_path("ws/socket.io/"); + url.set_query(req.uri().query()); + + helper::ws::manage_connection(req, pl, url).await +} + +#[instrument(skip(payload))] +async fn openwebui_forward( + req: HttpRequest, + payload: web::Payload, + method: Method, + peer_addr: Option, + owui_cookie: Option>, + client: web::Data, +) -> Result { + let owui_cookie = match owui_cookie { + Some(cookie) => cookie.into_inner(), + None => { + return Err(BridgeError::Unauthorized( + "OWUI cookie not found".to_string(), + )); + } + }; + + let mut url = Url::from_str(&make_forward_url("http", &owui_cookie.subject, false))?; + let path = req.path(); + url.set_path(path); + + if WHITELIST_ENDPOINTS.contains(path) + && let Ok(ref mut p) = url.path_segments_mut() + { + p.push(""); + } + + url.set_query(req.uri().query()); + + helper::forwarding::forward(req, payload, method, peer_addr, client, url, None, false).await +} + +#[instrument(skip(payload))] +async fn moleviewer_forward( + req: HttpRequest, + payload: web::Payload, + method: Method, + peer_addr: Option, + owui_cookie: Option>, + client: web::Data, +) -> Result { + let owui_cookie = match owui_cookie { + Some(cookie) => cookie.into_inner(), + None => { + return Err(BridgeError::Unauthorized( + "OWUI cookie not found".to_string(), + )); + } + }; + + let mut url = Url::from_str(&make_forward_url("http", &owui_cookie.subject, true))?; + let path = req.path(); + url.set_path(path); + url.set_query(req.uri().query()); + + helper::forwarding::forward(req, payload, method, peer_addr, client, url, None, false).await +} + +#[inline] +pub(crate) fn make_forward_url(protocol: &str, subject: &str, mole: bool) -> String { + let namespace = *OWUI_NAMESPACE; + let port = if mole { MOLE_VIEW_PORT } else { OWUI_PORT }; + format!("{protocol}://u{subject}-openwebui.{namespace}.svc.cluster.local:{port}") +} + +pub fn config_openwebui(cfg: &mut web::ServiceConfig) { + cfg.service(openwebui_ws) + .default_service(web::to(openwebui_forward)); +} + +pub fn config_moleviewer(cfg: &mut web::ServiceConfig) { + cfg.default_service(web::to(moleviewer_forward)); +} + +#[cfg(test)] +mod tests { + #[test] + fn test_make_forward_url() { + // kind of a silly test; can be used as a contract on how url is set + let protocol = "http"; + let subject = "test-subject"; + let namespace = "openwebui"; + let port = "8080"; + let expected_url = + format!("{protocol}://u{subject}-openwebui.{namespace}.svc.cluster.local:{port}"); + assert_eq!( + super::make_forward_url(protocol, subject, false), + expected_url + ); + } +} diff --git a/src/web/route/portal/group_admin/mod.rs b/src/web/route/portal/group_admin/mod.rs index 65726ffc..06f690fc 100644 --- a/src/web/route/portal/group_admin/mod.rs +++ b/src/web/route/portal/group_admin/mod.rs @@ -19,7 +19,7 @@ use crate::{ Database, models::{ AdminTab, AdminTabs, BridgeCookie, GROUP, Group, ModifyUser, NotebookStatusCookie, - USER, User, UserGroupMod, UserPortalRep, UserType, + OWUICookie, USER, User, UserGroupMod, UserPortalRep, UserType, }, mongo::DB, }, @@ -48,6 +48,7 @@ pub(super) async fn group( req: HttpRequest, context: Data, nsc: Option>, + oc: Option>, db: Data<&DB>, subject: Option>, ) -> Result { @@ -110,6 +111,14 @@ pub(super) async fn group( helper::add_token_exp_to_tera(&mut ctx, token); } + #[cfg(feature = "openwebui")] + if let Some(owui_cookie) = oc { + use crate::config::CONFIG; + + ctx.insert("openwebui", &owui_cookie.subject); + ctx.insert("owui_url", &CONFIG.openweb_url); + } + // add notebook tab if user has a notebook subscription #[cfg(feature = "notebook")] let nb_cookies = notebook_bookkeeping(&user, nsc, &mut bridge_cookie, &mut ctx, subs).await?; diff --git a/src/web/route/portal/helper.rs b/src/web/route/portal/helper.rs index eabccb7b..db565956 100644 --- a/src/web/route/portal/helper.rs +++ b/src/web/route/portal/helper.rs @@ -111,8 +111,11 @@ where user_notebook.status = nsc.status; } None => { + use crate::kube::NOTEBOOK_NAMESPACE; + let pvc = notebook_helper::make_notebook_volume_name(&user._id.to_string()); - if let Ok(true) = KubeAPI::::check_pvc_exists(&pvc).await { + if let Ok(true) = KubeAPI::::check_pvc_exists(&pvc, *NOTEBOOK_NAMESPACE).await + { bc.config = Some(crate::db::models::Config { notebook_persist_pvc: Some(true), }); @@ -124,13 +127,20 @@ where // notebook_cookie isn't there either... if let Some(nb_start) = &user.notebook { let sub = notebook_helper::make_notebook_name(&user._id.to_string()); - match KubeAPI::::check_pod_running(&(sub.clone() + "-0")).await { + match KubeAPI::::check_pod_running( + &(sub.clone() + "-0"), + *NOTEBOOK_NAMESPACE, + ) + .await + { Ok(running) => { if running { user_notebook.status = "Ready".to_string(); ctx.insert("notebook", &user_notebook); - let ip = KubeAPI::::get_pod_ip(&(sub + "-0")).await?; + let ip = + KubeAPI::::get_pod_ip(&(sub + "-0"), *NOTEBOOK_NAMESPACE) + .await?; let notebook_cookie = NotebookCookie { subject: user._id.to_string(), ip, diff --git a/src/web/route/portal/mod.rs b/src/web/route/portal/mod.rs index 5046ea20..c139a885 100644 --- a/src/web/route/portal/mod.rs +++ b/src/web/route/portal/mod.rs @@ -15,7 +15,8 @@ use tracing::instrument; #[cfg(feature = "observe")] use crate::web::helper::observability_post; use crate::{ - auth::{COOKIE_NAME, NOTEBOOK_COOKIE_NAME, NOTEBOOK_STATUS_COOKIE_NAME}, + auth::{COOKIE_NAME, NOTEBOOK_COOKIE_NAME, NOTEBOOK_STATUS_COOKIE_NAME, OWUI_COOKIE_NAME}, + config::CONFIG, db::{ Database, models::{BridgeCookie, USER, User, UserPortalRep, UserType}, @@ -60,7 +61,33 @@ async fn index(data: Option>, db: Data<&DB>) -> Result>() }); @@ -137,7 +164,7 @@ async fn search_by_email( }; let email = urlencoding::decode(email).map_err(|e| { tracing::error!("{}", e); - BridgeError::GeneralError(format!("Error parsing query string: {}", e)) + BridgeError::GeneralError(format!("Error parsing query string: {e}")) })?; let res = match db.search_users(&email, USER, PhantomData::).await { @@ -145,10 +172,7 @@ async fn search_by_email( Err(e) => match e { BridgeError::RecordSearchError(_) => { return Ok(HttpResponse::BadRequest() - .append_header(( - HTMX_ERROR_RES, - format!("No email found for {}", email), - )) + .append_header((HTMX_ERROR_RES, format!("No email found for {email}"))) .finish()); } _ => { @@ -214,6 +238,15 @@ async fn logout(#[cfg(feature = "observe")] req: HttpRequest) -> HttpResponse { .finish(); notebook_cookie.make_removal(); + let mut openwebui_cookie = Cookie::build(OWUI_COOKIE_NAME, "") + .domain(&CONFIG.bridge_url) + .same_site(SameSite::Lax) + .path("/") + .http_only(true) + .secure(true) + .finish(); + openwebui_cookie.make_removal(); + let mut notebook_status_cookie = Cookie::build(NOTEBOOK_STATUS_COOKIE_NAME, "") .same_site(SameSite::Strict) .path("/") @@ -236,6 +269,7 @@ async fn logout(#[cfg(feature = "observe")] req: HttpRequest) -> HttpResponse { .cookie(cookie_remove) .cookie(notebook_cookie) .cookie(notebook_status_cookie) + .cookie(openwebui_cookie) .finish() } diff --git a/src/web/route/portal/system_admin/mod.rs b/src/web/route/portal/system_admin/mod.rs index f2cd0238..5be97483 100644 --- a/src/web/route/portal/system_admin/mod.rs +++ b/src/web/route/portal/system_admin/mod.rs @@ -21,7 +21,8 @@ use crate::{ Database, models::{ AdminTab, AdminTabs, BridgeCookie, GROUP, Group, GroupForm, GroupPortalRep, - NotebookStatusCookie, USER, User, UserDeleteForm, UserForm, UserPortalRep, UserType, + NotebookStatusCookie, OWUICookie, USER, User, UserDeleteForm, UserForm, UserPortalRep, + UserType, }, mongo::DB, }, @@ -51,6 +52,7 @@ pub(super) async fn system( context: Data, subject: Option>, nsc: Option>, + oc: Option>, db: Data<&DB>, ) -> Result { // get the subject id from middleware @@ -111,6 +113,14 @@ pub(super) async fn system( helper::add_token_exp_to_tera(&mut ctx, token); } + #[cfg(feature = "openwebui")] + if let Some(owui_cookie) = oc { + use crate::config::CONFIG; + + ctx.insert("openwebui", &owui_cookie.subject); + ctx.insert("owui_url", &CONFIG.openweb_url); + } + // add notebook tab if user has a notebook subscription #[cfg(feature = "notebook")] let nb_cookies = notebook_bookkeeping(&user, nsc, &mut bridge_cookie, &mut ctx, subs).await?; @@ -183,7 +193,7 @@ async fn system_create_group( // TODO: check if group already exists, and not rely one dup key from DB let result = helper::log_with_level!(db.insert(group, GROUP).await, error); let content = match result { - Ok(r) => format!("

Group created with id: {}

", r), + Ok(r) => format!("

Group created with id: {r}

"), Err(e) if e.to_string().contains("dup key") => { return Ok(HttpResponse::BadRequest() .append_header(( diff --git a/src/web/route/portal/user.rs b/src/web/route/portal/user.rs index 58aca56d..a2fe8f98 100644 --- a/src/web/route/portal/user.rs +++ b/src/web/route/portal/user.rs @@ -14,7 +14,9 @@ use crate::{ auth::COOKIE_NAME, db::{ Database, - models::{BridgeCookie, GROUP, Group, NotebookStatusCookie, USER, User, UserType}, + models::{ + BridgeCookie, GROUP, Group, NotebookStatusCookie, OWUICookie, USER, User, UserType, + }, mongo::DB, }, errors::{BridgeError, Result}, @@ -31,6 +33,7 @@ pub(super) async fn user( req: HttpRequest, subject: Option>, nsc: Option>, + oc: Option>, db: Data<&DB>, ) -> Result { // get the subject id from middleware @@ -86,6 +89,7 @@ pub(super) async fn user( data, context, nsc, + oc, &mut bridge_cookie, helper::add_token_exp_to_tera ) diff --git a/src/web/route/portal/user_htmx.rs b/src/web/route/portal/user_htmx.rs index 708f98a1..dd10ce73 100644 --- a/src/web/route/portal/user_htmx.rs +++ b/src/web/route/portal/user_htmx.rs @@ -5,7 +5,7 @@ use actix_web::{ use tera::{Context, Tera}; use crate::{ - db::models::{BridgeCookie, NotebookStatusCookie, User}, + db::models::{BridgeCookie, NotebookStatusCookie, OWUICookie, User}, errors::Result, web::services::CATALOG, }; @@ -45,6 +45,7 @@ impl<'p> Profile<'p> { tera: Data, context: Data, nsc: Option>, + oc: Option>, bc: &mut BridgeCookie, t_exp: impl FnOnce(&mut Context, &str), ) -> Result<(String, Option<[Cookie; 2]>)> { @@ -79,6 +80,14 @@ impl<'p> Profile<'p> { context.insert("resources", &resources); } + #[cfg(feature = "openwebui")] + if let Some(owui_cookie) = oc { + use crate::config::CONFIG; + + context.insert("openwebui", &owui_cookie.subject); + context.insert("owui_url", &CONFIG.openweb_url); + } + #[cfg(feature = "notebook")] let nb_cookies = notebook_bookkeeping(self.user, nsc, bc, &mut context, self.subscriptions.clone()) diff --git a/src/web/route/proxy/mod.rs b/src/web/route/proxy/mod.rs index 58d9fc6d..7478d8cb 100644 --- a/src/web/route/proxy/mod.rs +++ b/src/web/route/proxy/mod.rs @@ -47,7 +47,7 @@ async fn forward( new_url.set_path(path); new_url.set_query(req.uri().query()); - helper::forwarding::forward(req, payload, method, peer_addr, client, new_url, None).await + helper::forwarding::forward(req, payload, method, peer_addr, client, new_url, None, true).await } pub fn config_proxy(cfg: &mut web::ServiceConfig) { diff --git a/src/web/route/resource/mod.rs b/src/web/route/resource/mod.rs index 6aad19bf..4bde19bf 100644 --- a/src/web/route/resource/mod.rs +++ b/src/web/route/resource/mod.rs @@ -62,7 +62,7 @@ async fn resource_http( bridge_cookie.token = Some(token); let content = serde_json::to_string(&bridge_cookie).map_err(|e| { - BridgeError::GeneralError(format!("Could not serialize bridge cookie: {}", e)) + BridgeError::GeneralError(format!("Could not serialize bridge cookie: {e}")) })?; Some( Cookie::build(COOKIE_NAME, content) @@ -89,6 +89,7 @@ async fn resource_http( client, new_url, updated_cookie, + false, ) .await } diff --git a/templates/components/owui/owui.html b/templates/components/owui/owui.html new file mode 100644 index 00000000..2cce013f --- /dev/null +++ b/templates/components/owui/owui.html @@ -0,0 +1,18 @@ +
+
+

Your Open Web UI

+
+

You currently have a owui running.

+
+ + Open Web UI + +
+
+
+
diff --git a/templates/pages/nav_base.html b/templates/pages/nav_base.html index cb4e2e3d..91071d58 100644 --- a/templates/pages/nav_base.html +++ b/templates/pages/nav_base.html @@ -123,6 +123,23 @@ {% endif %} + {% if openwebui %} + +
+ +

Open Web UI

+
+ + {% endif %} {% if user_type == "system" %}
@@ -204,6 +221,9 @@ {% if notebook %} Notebook {% endif %} + {% if openwebui %} + Open Web UI + {% endif %} {% if user_type == "system" %} System Management diff --git a/templates/pages/portal_user.html b/templates/pages/portal_user.html index b6c96367..7c256c71 100644 --- a/templates/pages/portal_user.html +++ b/templates/pages/portal_user.html @@ -25,6 +25,12 @@
{% endif %} +{% if openwebui %} + +{% endif %} + {% if user_type == "group" %}