diff options
| author | Somhairle H. Marisol <[email protected]> | 2026-09-17 14:32:37 +0800 |
|---|---|---|
| committer | Somhairle H. Marisol <[email protected]> | 2026-09-17 14:32:37 +0800 |
| commit | 5c0ba37eda80d39e6ceca59bb1d5f4942f858995 (patch) | |
| tree | 948723f9cedf7ccb0707fa6ee516bd30fe20fd10 /server | |
| download | strategy-lab-5c0ba37eda80d39e6ceca59bb1d5f4942f858995.tar.gz | |
chore: establish Strategy Lab source baseline (development, not release)
Diffstat (limited to 'server')
| -rw-r--r-- | server/Cargo.lock | 1984 | ||||
| -rw-r--r-- | server/Cargo.toml | 30 | ||||
| -rw-r--r-- | server/src/admin.rs | 387 | ||||
| -rw-r--r-- | server/src/ai.rs | 319 | ||||
| -rw-r--r-- | server/src/auth.rs | 713 | ||||
| -rw-r--r-- | server/src/config.rs | 65 | ||||
| -rw-r--r-- | server/src/datasets.rs | 244 | ||||
| -rw-r--r-- | server/src/db.rs | 146 | ||||
| -rw-r--r-- | server/src/error.rs | 145 | ||||
| -rw-r--r-- | server/src/jobs.rs | 857 | ||||
| -rw-r--r-- | server/src/main.rs | 667 | ||||
| -rw-r--r-- | server/src/projects.rs | 219 | ||||
| -rw-r--r-- | server/src/runs.rs | 281 | ||||
| -rw-r--r-- | server/src/state.rs | 55 | ||||
| -rw-r--r-- | server/src/store.rs | 206 | ||||
| -rw-r--r-- | server/src/util.rs | 138 | ||||
| -rw-r--r-- | server/src/worker.rs | 482 |
17 files changed, 6938 insertions, 0 deletions
diff --git a/server/Cargo.lock b/server/Cargo.lock new file mode 100644 index 0000000..41abaa4 --- /dev/null +++ b/server/Cargo.lock @@ -0,0 +1,1984 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + +[[package]] +name = "android_system_properties" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" +dependencies = [ + "libc", +] + +[[package]] +name = "argon2" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" +dependencies = [ + "base64ct", + "blake2", + "cpufeatures 0.2.17", + "password-hash", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "axum-macros", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-macros" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "base64ct" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "blake2" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" +dependencies = [ + "digest", +] + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3eb0f42d6c360dc3f8a821f6bf2fdea7f72bfd36b3076eb0e6d1e9e0752fff4" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.1", + "rand_core 0.10.1", +] + +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "js-sys", + "num-traits", + "serde", + "wasm-bindgen", + "windows-link", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", + "subtle", +] + +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + +[[package]] +name = "find-msvc-tools" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e0f1c7c3a72c66fd80abe965175f7523475c0489a87d3ff9d6e8c87d87a9d2d" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "wasi", + "wasm-bindgen", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "r-efi", + "rand_core 0.10.1", + "wasm-bindgen", +] + +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashlink" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af" +dependencies = [ + "hashbrown", +] + +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "hyper", + "ipnet", + "libc", + "percent-encoding", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "js-sys" +version = "0.3.105" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce57d20d1ea864ce2ac172ab472d409214f4fd359f0b2a2775abdf522e2af99e" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "lru-slab" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mio" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "password-hash" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "346f04948ba92c43e8469c1ee6736c7563d71012b17d40745260fe106aac2166" +dependencies = [ + "base64ct", + "rand_core 0.6.4", + "subtle", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "pkg-config" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" + +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quinn" +version = "0.11.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9746dbde176634f4f2f1faf2404e30a31b2bc1e9cafb5329c95d8177a18c9fc" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand 0.10.2", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c" +dependencies = [ + "libc", + "rand_chacha", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rusqlite" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls" +version = "0.23.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + +[[package]] +name = "strategy-lab-server" +version = "0.1.0" +dependencies = [ + "argon2", + "axum", + "chrono", + "hex", + "http-body-util", + "rand 0.8.8", + "reqwest", + "rusqlite", + "serde", + "serde_json", + "sha2", + "tempfile", + "tokio", + "tower", + "tracing", + "tracing-subscriber", + "uuid", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.3", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c85f2c3ef0b1cd58b36682f4b17aaa995f0e5db534d85692b4903abce21f67" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + +[[package]] +name = "uuid" +version = "1.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aecb87a33d3b0c5e3b7aa46336eaf486cffafbd281b195e4c8b80d50df2351bf" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.78" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ef4c5d3d2cdf5c54f4231181768f5510842e350db025faf1f7163b1030ed928" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a690d511e3c1a8b3a55e33511e3c2c00c78415cd23650f32b808627f5696b9ed" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "411e4887f0071ef2d2164a9d5fdf2d20efbef78fccd3a78b0c10a1dc5295e48a" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 3.0.5", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81941cd78d0c92026c33e5e01312845a4cb1e9af3407f9134b100dd03144103e" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.105" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fbddc4a036f00ec4f18c83445bd3115cb306a91da554919a099d9222fe4a7f8" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33811428bee40dbceb6d545e95754741d17a6aef9a4849f0fd62e2ba4f412a78" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", + "synstructure", +] + +[[package]] +name = "zerocopy" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d35102a9f36d089ccae9e4c6802bc118be4487b80aaffc0ab4e0cf5ce92d2873" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "146c01f5ab44258da43cf276c74a2763db2ff3969c9c652c3f2de07041d0b2bc" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", + "synstructure", +] + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/server/Cargo.toml b/server/Cargo.toml new file mode 100644 index 0000000..88e512f --- /dev/null +++ b/server/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "strategy-lab-server" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "strategy-lab-server" +path = "src/main.rs" + +[dependencies] +axum = { version = "0.8", features = ["json", "http1", "macros"] } +tokio = { version = "1", features = ["full"] } +rusqlite = { version = "0.32", features = ["bundled"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +uuid = { version = "1", features = ["v4"] } +argon2 = "0.5" +rand = "0.8" +sha2 = "0.10" +hex = "0.4" +chrono = { version = "0.4", features = ["serde"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +tower = "0.5" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[dev-dependencies] +tempfile = "3" +http-body-util = "0.1" +tower = { version = "0.5", features = ["util"] } diff --git a/server/src/admin.rs b/server/src/admin.rs new file mode 100644 index 0000000..d6a861e --- /dev/null +++ b/server/src/admin.rs @@ -0,0 +1,387 @@ +use serde_json::json; +use axum::Json; + +use crate::auth::{audit, hash_password, make_admin_token, AuthUser}; +use crate::error::{AppError, AppResult}; +use crate::state::Cx; +use crate::util::now_iso; + +async fn assert_admin(_cx: &Cx, auth: &AuthUser) -> AppResult<()> { + if auth.role != "admin" { + return Err(AppError::forbidden("admin only")); + } + Ok(()) +} + +fn user_json(id: &str, email: &str, name: &str, role: &str, active: i64, ai: i64, limit: i64, created: String) -> serde_json::Value { + json!({ + "id": id, "email": email, "name": name, "role": role, + "active": active != 0, "ai_enabled": ai != 0, + "daily_run_limit": limit, "created_at": created, + }) +} + +pub async fn users(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> { + assert_admin(&cx, &auth).await?; + let items: Vec<serde_json::Value> = cx.with_db(|db| { + let mut st = db.prepare("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users ORDER BY created_at ASC")?; + let v: Vec<serde_json::Value> = st.query_map([], |r| Ok(user_json( + &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?, + r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?, + )))?.collect::<Result<_, _>>()?; + Ok::<_, AppError>(v) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +#[derive(serde::Deserialize)] +pub struct AdminUserPatch { + pub active: Option<bool>, + pub role: Option<String>, + pub daily_run_limit: Option<i64>, + pub ai_enabled: Option<bool>, +} + +pub async fn patch_user(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, body: Option<axum::Json<AdminUserPatch>>) -> AppResult<Json<serde_json::Value>> { + assert_admin(&cx, &auth).await?; + let axum::Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if let Some(role) = &r.role { + if !matches!(role.as_str(), "admin" | "member") { + return Err(AppError::bad("validation", "role must be admin|member")); + } + } + if let Some(n) = r.daily_run_limit { + if !(1..=2000).contains(&n) { + return Err(AppError::bad("validation", "daily_run_limit must be between 1 and 2000")); + } + } + cx.with_db(|db| -> AppResult<()> { + let tx = db.transaction()?; + let (trole, tactive): (String, i64) = tx.query_row( + "SELECT role, active FROM users WHERE id=?1", [&target], |row| Ok((row.get(0)?, row.get(1)?))) + .map_err(|_| AppError::not_found("user not found"))?; + let demoting = r.role.as_ref().map(|new| trole == "admin" && new != "admin").unwrap_or(false); + let disabling = r.active == Some(false); + // Last active admin protection, enforced atomically with the update. + if (demoting || disabling) && trole == "admin" && tactive != 0 { + let active_admins: i64 = tx.query_row("SELECT COUNT(*) FROM users WHERE role='admin' AND active=1", [], |row| row.get(0))?; + if active_admins <= 1 { + return Err(AppError::conflict("last_admin", "cannot demote or disable the last active admin")); + } + } + tx.execute( + "UPDATE users SET ai_enabled=COALESCE(?1,ai_enabled), daily_run_limit=COALESCE(?2,daily_run_limit), role=COALESCE(?3,role), active=COALESCE(?4,active) WHERE id=?5", + rusqlite::params![ + r.ai_enabled.map(|b| b as i64), + r.daily_run_limit, + &r.role, + r.active.map(|b| b as i64), + &target, + ])?; + // Disabling revokes every session and freezes that user's live runs, + // inside the same transaction as the account state flip. + if disabling { + tx.execute("DELETE FROM sessions WHERE user_id=?1", [&target]).ok(); + tx.execute( + "UPDATE runs SET status='cancelled', error='account disabled', finished_at=?1 WHERE user_id=?2 AND status IN ('queued','running')", + rusqlite::params![now_iso(), &target], + ).ok(); + } + tx.commit()?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "admin_user_update", &target, "ok").await; + let v = cx.with_db(|db| { + db.query_row("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users WHERE id=?1", [&target], |r| + Ok(user_json( + &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?, + r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?, + ))).map_err(|_| AppError::not_found("user not found")) + }).await?; + Ok(Json(v)) +} + +#[derive(serde::Deserialize)] +pub struct InvitationReq { + pub email: Option<String>, + pub role: Option<String>, + pub expires_hours: Option<i64>, +} + +pub async fn create_invitation(cx: Cx, auth: AuthUser, body: Option<axum::Json<InvitationReq>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> { + assert_admin(&cx, &auth).await?; + let axum::Json(r) = body.unwrap_or(axum::Json(InvitationReq { email: None, role: None, expires_hours: None })); + let role = r.role.unwrap_or_else(|| "member".to_string()); + // POC: invitations can only mint members; admins are bootstrapped offline. + // The invitation's role is stored in DB and registration ignores any + // client-supplied role input, so no escalation path exists. + if role != "member" { + return Err(AppError::bad("validation", "POC invitations can only create member role")); + } + let hours = r.expires_hours.unwrap_or(168); + if !(1..=336).contains(&hours) { + return Err(AppError::bad("validation", "expires_hours must be 1-336")); + } + let (token, expires) = make_admin_token(&cx, "invitations", None, hours, r.email.as_deref()).await?; + audit(&cx, Some(&auth.id), "invitation_issued", r.email.as_deref().unwrap_or("open-invite"), "ok").await; + Ok((axum::http::StatusCode::CREATED, Json(json!({ "token": token, "expires_at": expires })))) +} + +/// Sanitized invitations list: no token hashes, no secrets. +pub async fn list_invitations(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> { + assert_admin(&cx, &auth).await?; + let items: Vec<serde_json::Value> = cx.with_db(|db| { + let now = now_iso(); + let mut st = db.prepare("SELECT id,email,role,expires_at,used_by,created_at FROM invitations WHERE expires_at > ?1 ORDER BY created_at DESC")?; + let v = st.query_map([&now], |r| Ok(json!({ + "id": r.get::<_, String>(0)?, + "email": r.get::<_, Option<String>>(1)?, + "role": r.get::<_, String>(2)?, + "expires_at": r.get::<_, String>(3)?, + "used_by": r.get::<_, Option<String>>(4)?, + "created_at": r.get::<_, String>(5)?, + })))?.collect::<Result<Vec<_>, _>>()?; + Ok::<_, AppError>(v) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +pub async fn delete_invitation(cx: Cx, auth: AuthUser, axum::extract::Path(id): axum::extract::Path<String>) -> AppResult<Json<serde_json::Value>> { + assert_admin(&cx, &auth).await?; + cx.with_db(|db| -> AppResult<()> { + if db.execute("DELETE FROM invitations WHERE id=?1", [&id])? == 0 { + return Err(AppError::not_found("invitation not found")); + } + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "invitation_revoke", &id, "ok").await; + Ok(Json(json!({"ok": true}))) +} + +/// Admin-issued password reset token, hashed in DB, short-lived single-use. +/// The raw token is returned once for display; no fake email delivery. +pub async fn create_reset(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, _body: Option<axum::Json<serde_json::Value>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> { + assert_admin(&cx, &auth).await?; + cx.with_db(|db| { + db.query_row("SELECT 1 FROM users WHERE id=?1", [&target], |row| row.get::<_, i64>(0)) + .map_err(|_| AppError::not_found("user not found"))?; + Ok::<_, AppError>(()) + }).await?; + let hours = 2; + let (token, expires) = make_admin_token(&cx, "password_resets", Some(target.as_str()), hours, None).await?; + audit(&cx, Some(&auth.id), "admin_password_reset_issued", &target, "ok").await; + Ok((axum::http::StatusCode::CREATED, Json(json!({"reset_token": token, "expires_at": expires})))) +} + +/// Sanitized security audit listing: actor/action/target/time/status only. +/// Sensitive material never reaches this table (see auth::audit) and stored +/// targets are rendered trimmed, never with password/key/code content. +pub async fn audit_list(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> { + assert_admin(&cx, &auth).await?; + let items: Vec<serde_json::Value> = cx.with_db(|db| { + let mut st = db.prepare("SELECT ts,actor_id,action,target,status FROM audit ORDER BY seq DESC LIMIT 500")?; + let v = st.query_map([], |r| { + let ts: String = r.get(0)?; + let actor: Option<String> = r.get(1)?; + let action: String = r.get(2)?; + let target: Option<String> = r.get(3)?; + let status: String = r.get(4)?; + let items_json = json!({ + "ts": ts, "actor": actor, "action": action, + "target": target.unwrap_or_default(), "status": status, + }); + Ok(items_json) + })?.collect::<Result<Vec<_>, _>>()?; + Ok::<_, AppError>(v) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +/// Bootstrap the first admin from env. Idempotent: only inserts when no +/// admin account exists yet. +pub async fn bootstrap_admin(cx: &Cx) -> AppResult<()> { + let cfg = &cx.cfg; + let Some(email) = cfg.bootstrap_admin_email.clone() else { return Ok(()); }; + let Some(password) = cfg.bootstrap_admin_password.clone() else { return Ok(()); }; + let email = email.trim().to_lowercase(); + if email.is_empty() || !email.contains('@') || password.len() < 8 { + return Ok(()); + } + let exists: i64 = cx.with_db(|db| { + db.query_row("SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap_or(0) + }).await; + if exists > 0 { return Ok(()); } + let hash = hash_password(&password)?; + let id = crate::util::new_id(); + cx.with_db(|db| { + db.execute("INSERT OR IGNORE INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,'Administrator','admin',1,1,100,?4)", + rusqlite::params![&id, &email, &hash, now_iso()]).ok(); + }).await; + audit(cx, Some(&id), "bootstrap_admin", &id, "ok").await; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::state::AppState; + use axum::extract::{Path as P, State}; + use std::sync::Arc; + + fn handle(s: &Arc<AppState>) -> Cx { State(s.clone()) } + + fn admin_auth() -> AuthUser { + AuthUser { id: "admin-x".into(), email: "[email protected]".into(), role: "admin".into(), ai_enabled: true, daily_run_limit: 100, session_id: "sess-admin".into() } + } + + fn member_auth() -> AuthUser { + AuthUser { id: "member-x".into(), email: "[email protected]".into(), role: "member".into(), ai_enabled: false, daily_run_limit: 10, session_id: "sess-member".into() } + } + + fn test_state() -> Arc<AppState> { + let conn = rusqlite::Connection::open_in_memory().expect("in-memory db"); + crate::db::init_db(&conn).expect("schema"); + let cfg = crate::config::Config { + bind_addr: "127.0.0.1:0".into(), + canonical_origin: String::new(), + secure_cookies: false, + db_path: ":memory:".into(), + frontend_dir: "frontend/dist".into(), + data_dir: std::env::temp_dir().to_string_lossy().into_owned(), + worker_image: "test".into(), + fetch_timeout_secs: 1, + backtest_timeout_secs: 1, + run_concurrency: 1, + fetch_concurrency: 1, + session_hours: 24, + bootstrap_admin_email: None, + bootstrap_admin_password: None, + ai_base_url: "http://127.0.0.1:9".into(), + ai_model: "test-model".into(), + ai_daily_request_cap: 1, + ai_input_token_cap: 1, + ai_output_token_cap: 1, + ai_enabled_poc: false, + default_run_limit_per_day: 10, + version: "test".into(), + }; + Arc::new(AppState { + cfg, + db: tokio::sync::Mutex::new(conn), + run_sem: Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)), + }) + } + + /// Insert a user directly and return its id. + async fn seed_user(cx: &Cx, email: &str, role: &str, password: &str) -> String { + let hash = crate::auth::hash_password(password).unwrap(); + let uid = crate::util::new_id(); + cx.with_db(|db| { + db.execute( + "INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,?4,?5,1,0,10,?6)", + rusqlite::params![&uid, email, &hash, "Test User", role, now_iso()]).unwrap(); + }).await; + uid + } + + #[tokio::test] + async fn member_cannot_list_users_or_audit() { + let s = test_state(); + let u = users(handle(&s), member_auth()).await; + assert_eq!(u.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN); + let a = audit_list(handle(&s), member_auth()).await; + assert_eq!(a.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN); + let inv = create_invitation(handle(&s), member_auth(), None).await; + assert_eq!(inv.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn last_active_admin_cannot_be_disabled_or_demoted() { + let s = test_state(); + let aid = seed_user(&handle(&s), "[email protected]", "admin", "an-admin-passphrase").await; + let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await; + assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "disabling the last admin must be blocked"); + let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: None, role: Some("member".into()), daily_run_limit: None, ai_enabled: None }))).await; + assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "demoting the last admin must be blocked"); + // With a second active admin, disabling becomes legal. + let _ = seed_user(&handle(&s), "[email protected]", "admin", "another-passphrase-2").await; + let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await; + assert!(ok.is_ok()); + let out = ok.unwrap().0; + assert_eq!(out["active"], json!(false)); + } + + #[tokio::test] + async fn disabling_user_revokes_sessions_and_cancels_runs() { + let s = test_state(); + let aid = seed_user(&handle(&s), "[email protected]", "member", "a-member-passphrase").await; + handle(&s).with_db(|db| { + db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES ('sess-1',?1,?2,?3)", + rusqlite::params![&aid, now_iso(), crate::util::plus_hours(1)]).unwrap(); + db.execute( + "INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('pid',?1,'p','',0,?2,?2)", + rusqlite::params![&aid, now_iso()]).unwrap(); + db.execute( + "INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('vid','pid','code','h','m','manual',?1)", + [now_iso()]).unwrap(); + db.execute( + "INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('did',?1,'ds','{}','ready',?2,?2)", + rusqlite::params![&aid, now_iso()]).unwrap(); + db.execute( + "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('rid',?1,'pid','vid','did','running','{}',?2)", + rusqlite::params![&aid, now_iso()]).unwrap(); + }).await; + let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await; + assert!(ok.is_ok(), "member can be disabled"); + let blocked: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM runs WHERE id='rid' AND status='cancelled'", [], |r| r.get(0)).unwrap()).await; + assert_eq!(blocked, 1, "live runs must be cancelled"); + let gone: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM sessions WHERE user_id=?1", [&aid], |r| r.get(0)).unwrap()).await; + assert_eq!(gone, 0, "sessions must be revoked"); + } + + #[tokio::test] + async fn invitations_only_mint_members_and_are_hashed() { + let s = test_state(); + let res = create_invitation(handle(&s), admin_auth(), None).await.unwrap(); + assert_eq!(res.0, axum::http::StatusCode::CREATED); + let token = res.1.0["token"].as_str().unwrap().to_string(); + assert!(!token.contains("password")); + // The DB must hold only the sha256 of the token, never the raw token. + let hash = crate::util::sha256_hex(token.as_bytes()); + let hashed_rows: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&hash], |r| r.get(0)).unwrap()).await; + assert_eq!(hashed_rows, 1); + let raw_rows: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&token], |r| r.get(0)).unwrap()).await; + assert_eq!(raw_rows, 0, "raw tokens must never be stored"); + // Admin roles via invitation are rejected. + let offered = create_invitation(handle(&s), admin_auth(), Some(axum::Json(InvitationReq { email: None, role: Some("admin".into()), expires_hours: None }))).await; + assert_eq!(offered.err().map(|e| e.code).unwrap_or_default(), "validation"); + // Sanitized listing contains no token material. + let list = list_invitations(handle(&s), admin_auth()).await.unwrap().0; + let raw = serde_json::to_string(&list).unwrap(); + assert!(!raw.contains(&hash), "listing must not leak token hashes"); + } + + #[tokio::test] + async fn bootstrap_admin_is_idempotent_and_hashes_password() { + let mut cfg_state = test_state(); + { + let cfg = &mut Arc::get_mut(&mut cfg_state).unwrap().cfg; + cfg.bootstrap_admin_email = Some("[email protected] ".into()); + cfg.bootstrap_admin_password = Some("boot-admin-passphrase".into()); + } + let cx = handle(&cfg_state); + bootstrap_admin(&cx).await.unwrap(); + bootstrap_admin(&cx).await.unwrap(); + let count: i64 = cx.with_db(|db| db.query_row( + "SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await; + assert_eq!(count, 1, "bootstrap must not duplicate admins"); + let hash: String = cx.with_db(|db| db.query_row( + "SELECT password_hash FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await; + assert!(hash.starts_with("$argon2"), "bootstrap password must be Argon2 hashed"); + } +} diff --git a/server/src/ai.rs b/server/src/ai.rs new file mode 100644 index 0000000..271fa76 --- /dev/null +++ b/server/src/ai.rs @@ -0,0 +1,319 @@ +use axum::extract::Path; +use axum::http::StatusCode; +use axum::Json; +use serde_json::{json, Value}; + +use crate::auth::{audit, AuthUser}; +use crate::error::{AppError, AppResult}; +use crate::util::new_id; +use crate::util::now_iso; +use crate::util::sha256_hex; + +use crate::state::Cx; + +const ASSIST_TIMEOUT_SECS: u64 = 90; +pub const MAX_INSTRUCTION_CHARS: usize = 2000; + +pub async fn list_ai_usage(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> { + let (items, totals_in, totals_out, n) = cx.with_db(|db| -> AppResult<_> { + let mut items = Vec::new(); + { + let mut st = db.prepare("SELECT ts,kind,input_tokens,output_tokens FROM ai_usage WHERE user_id=?1 ORDER BY ts DESC LIMIT 500")?; + let mut rows = st.query([auth.id.clone()])?; + while let Some(r) = rows.next()? { items.push(json!({ + "ts": r.get::<_, String>(0)?, + "kind": r.get::<_, String>(1)?, + "input_tokens": r.get::<_, i64>(2)?, + "output_tokens": r.get::<_, i64>(3)?, + })); + } + } + let (i, o, n): (i64, i64, i64) = db.query_row( + "SELECT COALESCE(SUM(input_tokens),0), COALESCE(SUM(output_tokens),0), COUNT(*) FROM ai_usage WHERE user_id=?1", + [&auth.id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?; + Ok((items, i, o, n)) + }).await?; + Ok(Json(json!({ + "items": items, + "totals": {"requests": n, "input_tokens": totals_in, "output_tokens": totals_out, "internal_poc": true}, + }))) +} + +pub async fn assist(cx: Cx, auth: AuthUser, body: Option<Json<serde_json::Value>>) -> AppResult<Json<Value>> { + let axum::Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let project_id = j.get("project_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "project_id required"))?.to_string(); + let instruction = j.get("instruction").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "instruction required"))?.to_string(); + let expected = j.get("expected_generation").and_then(|v| v.as_i64()).ok_or_else(|| AppError::bad("validation", "expected_generation required"))?; + if instruction.trim().is_empty() || instruction.len() > MAX_INSTRUCTION_CHARS { + return Err(AppError::bad("validation", "instruction required (max 2000 chars)")); + } + if !cx.cfg.ai_enabled_poc { + return Err(AppError::forbidden("AI assist is not enabled (internal POC flag off)")); + } + if !auth.ai_enabled { + return Err(AppError::forbidden("AI not enabled on this account; contact the administrator")); + } + let (draft, gen): (String, i64) = cx.with_db(|db| { + db.query_row("SELECT draft_code,draft_generation FROM projects WHERE id=?1 AND user_id=?2", + rusqlite::params![&project_id, &auth.id], |r| Ok((r.get(0)?, r.get(1)?))) + .map_err(|_| AppError::not_found("project not found")) + }).await?; + if gen != expected { + return Err(AppError::conflict("stale_generation", "draft changed; expected generation mismatch")); + } + // daily request budget, measured from the real ledger + let key = std::env::var("OPENCODE_GO_API_KEY").map_err(|_| { + AppError::new(StatusCode::INTERNAL_SERVER_ERROR, "ai_not_configured", "[internal] AI provider key not configured in environment") + })?; + let (used, requests): (i64, i64) = cx.with_db(|db| { + let used: i64 = db.query_row("SELECT COUNT(*) FROM ai_requests WHERE user_id=?1 AND created_at LIKE ?2", + rusqlite::params![&auth.id, format!("{}%", chrono::Utc::now().format("%Y-%m-%d").to_string())], |r| r.get(0)).unwrap_or(0); + let reqs = db.query_row("SELECT COUNT(*) FROM ai_requests WHERE user_id=?1", [&auth.id], |r| r.get(0)).unwrap_or(0); + Ok::<_, AppError>((used, reqs)) + }).await?; + let _ = requests; + if used >= cx.cfg.ai_daily_request_cap { + return Err(AppError::new(StatusCode::TOO_MANY_REQUESTS, "ai_budget", "daily AI request budget reached")); + } + + let dataset_summary = summarize_dataset(&cx, &auth.id, &project_id).await?; + let system_prompt = "You are a strategy editor for a Backtrader POC. Propose changes to the user strategy code. Reply with one fenced ```python block containing the FULL proposed module, plus a short explanation outside the block. Never fabricate data. Do not execute anything."; + // Scope caveat (kept explicit): this integration is an internal POC providing + // strategy coding help for own use only; no production use and no commercial + // licensing grant is claimed for the upstream model service. + let user_prompt = format!("Change requested: {instruction}\n\nAvailable dataset fields (indicator warmup is your code's responsibility): {dataset_summary}\n\nCurrent full strategy source:\n{draft}"); + let body = json!({ + "model": cx.cfg.ai_model, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ], + "max_tokens": cx.cfg.ai_output_token_cap, + "temperature": 0.4, + }); + let id = new_id(); + let ts = now_iso(); + cx.with_db(|db| -> AppResult<()> { + db.execute("INSERT INTO ai_requests (id,user_id,project_id,instruction,status,base_generation,base_code_hash,created_at) VALUES (?1,?2,?3,?4,'pending',?5,?6,?7)", + rusqlite::params![&id, &auth.id, &project_id, &instruction, gen, sha256_hex(draft.as_bytes()), &ts])?; + Ok(()) + }).await?; + + let url = format!("{}/chat/completions", cx.cfg.ai_base_url.trim_end_matches('/')); + let client = reqwest::Client::builder().timeout(std::time::Duration::from_secs(ASSIST_TIMEOUT_SECS)).build() + .map_err(|_| AppError::internal("http client unavailable"))?; + let session_id = ai_session_id(&auth.id, &project_id); + let resp = client.post(url) + .bearer_auth(&key) + .header("Content-Type", "application/json") + // Honest app identity for the opencode go gateway (custom coding agents + // are an explicitly supported use; no other client identity is claimed). + .header("User-Agent", "strategy-lab-coding-assistant/0.1 (internal POC, strategy coding help only)") + // Stable per (account, project) conversation id so the upstream can + // reuse one session context; never random per request, never the key. + .header("x-opencode-session", session_id) + .json(&body).send().await; + + let resp: Value = match resp { + Ok(r) if r.status().is_success() => r.json().await.unwrap_or(Value::Null), + Ok(r) => { + let status = r.status(); + let text = r.text().await.unwrap_or_default(); + audit(&cx, Some(&auth.id), "ai_error", &id, "fail").await; + return Err(AppError::new(StatusCode::BAD_GATEWAY, "ai_upstream_error", format!("model call failed ({status})")) + .with_details(json!({"status": status.to_string(), "internal": text.chars().take(500).collect::<String>()}))); + } + Err(e) => { + audit(&cx, Some(&auth.id), "ai_error", &id, "error").await; + // startup must survive an unreachable provider; user sees a bounded error + return Err(AppError::new(StatusCode::BAD_GATEWAY, "ai_unreachable", format!("model endpoint unreachable: {e}"))); + } + }; + let usage_in: i64 = resp.get("usage").and_then(|u| u.get("prompt_tokens")).and_then(|v| v.as_i64()).unwrap_or(0); + let usage_out: i64 = resp.get("usage").and_then(|u| u.get("completion_tokens")).and_then(|v| v.as_i64()).unwrap_or(0); + if usage_in > cx.cfg.ai_input_token_cap { + return record_failure(&cx, &id, json!({"code": "ai_input_too_large", "input_tokens": usage_in}), "fail").await; + } + let choice_content = resp.get("choices").and_then(|c| c.get(0)).and_then(|c| c.get("message")) + .and_then(|m| m.get("content")).and_then(|c| c.as_str()).unwrap_or("").to_string(); + let proposed_opt = extract_code_block(&choice_content); + let explanation: String = { + let without = strip_code_blocks(&choice_content); + if without.is_empty() { "Model returned code without explanation.".to_string() } else { without } + }; + let Some(proposed) = proposed_opt else { + return record_failure(&cx, &id, json!({"code": "ai_no_code", "message": "model response contained no parseable full code block"}), "fail").await; + }; + let diff = crate::util::unified_diff(&draft, &proposed); + let usage = json!({"input_tokens": usage_in, "output_tokens": usage_out, "measured": true}); + cx.with_db(|db| -> AppResult<()> { + db.execute("UPDATE ai_requests SET status='succeeded', model=?1, explanation=?2, proposed_code=?3, diff=?4, usage=?5 WHERE id=?6", + rusqlite::params![cx.cfg.ai_model.clone(), explanation, &proposed, &diff, usage.to_string(), &id])?; + db.execute("INSERT INTO ai_usage (id,user_id,request_id,ts,kind,input_tokens,output_tokens) VALUES (?1,?2,?3,?4,'request',?5,?6)", + rusqlite::params![new_id(), &auth.id, &id, now_iso(), usage_in, usage_out])?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "ai_assist", &id, "ok").await; + Ok(Json(json!({ + "id": id, + "model": cx.cfg.ai_model, + "explanation": explanation, + "proposed_code": proposed, + "diff": diff, + "base_generation": gen, + "status": "succeeded", + "usage": usage, + }))) +} + +/// Stable AI gateway session id scoped to owner id + project UUID. +/// Deterministic so each (owner, project) pair always reuses one upstream +/// session; it contains no secrets (no API key material) and no email. +pub fn ai_session_id(owner_id: &str, project_id: &str) -> String { + format!("sl-strategy-lab-{}", sha256_hex(format!("{owner_id}:{project_id}").as_bytes())) +} + +async fn record_failure(cx: &Cx, id: &str, err: Value, status: &str) -> AppResult<Json<Value>> { + cx.with_db(|db| { + db.execute("UPDATE ai_requests SET status='failed', error=?1 WHERE id=?2", rusqlite::params![err.to_string(), id]).ok(); + }).await; + audit(cx, None, "ai_failure", id, status).await; + Err(AppError::bad("ai_failure", "AI request failed; see usage ledger")) +} + +fn strip_code_blocks(s: &str) -> String { + let mut out = String::new(); + let mut inn = false; + for line in s.lines() { + if line.contains("```python") || line.contains("```") { inn = !inn; continue } + if !inn { out.push_str(line); out.push('\n'); } + } + out.trim().to_string() +} + +pub fn extract_code_block(s: &str) -> Option<String> { + let mark = "```"; + let mut in_block = false; + let mut block = String::new(); + for l in s.lines() { + if !in_block && l.trim().starts_with(mark) { in_block = true; block.clear(); continue; } + if in_block && l.trim().starts_with(mark) { if !block.trim().is_empty() { return Some(block); } in_block = false; block.clear(); continue; } + if in_block { block.push_str(l); block.push('\n'); } + } + Some(block).filter(|b| !b.trim().is_empty()) +} + +/// Schema of the most recent ready dataset actually used by runs of this project. +async fn summarize_dataset(cx: &Cx, user_id: &str, project_id: &str) -> AppResult<String> { + let q: Option<String> = cx.with_db(|db| { + db.query_row("SELECT d.manifest FROM datasets d JOIN runs r ON r.dataset_id=d.id WHERE r.project_id=?1 AND d.user_id=?2 AND d.status='ready' ORDER BY r.created_at DESC LIMIT 1", + rusqlite::params![project_id, user_id], |r| r.get::<_, String>(0)).ok() + }).await; + let Some(m) = q else { return Ok("No dataset ready in this project yet: columns unknown.".to_string()); }; + let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null); + let mut cols = Vec::new(); + if let Some(objs) = mj.get("objects").and_then(|o| o.as_array()) { + for o in objs.iter() { + if let Some(c) = o.get("columns") { + if let Some(arr) = c.as_array() { + let cur: Vec<String> = arr.iter().filter_map(|v| v.as_str().map(String::from)).collect(); + if cur.len() > cols.len() { cols = cur; } + } + } + } + } + Ok(if cols.is_empty() { "Dataset ready but columns not enumerated.".to_string() } else { cols.join(", ") }) +} + +/// Accept a proposal and create a new draft and a version based on the AI proposal. +pub async fn accept(cx: Cx, auth: AuthUser, path: Path<(String,)>, body: Option<Json<serde_json::Value>>) -> AppResult<Json<Value>> { + let (aid,) = path.0; + let axum::Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let expected = j.get("expected_generation").and_then(|v| v.as_i64()).ok_or_else(|| AppError::bad("validation", "expected_generation required"))?; + let (user_id, project_id, proposed, base_gen, base_hash): (String, String, Option<String>, i64, Option<String>) = cx.with_db(|db| { + db.query_row("SELECT user_id,project_id,proposed_code,base_generation,base_code_hash FROM ai_requests WHERE id=?1", + [&aid], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))) + .map_err(|_| AppError::not_found("AI request not found")) + }).await?; + if user_id != auth.id { return Err(AppError::not_found("AI request not found")); } + let Some(proposed) = proposed else { return Err(AppError::bad("ai_not_ready", "AI request failed or has no code to accept")); }; + let p: Value = cx.with_db(|db| -> AppResult<Value> { + let (draft, gen): (String, i64) = db.query_row("SELECT draft_code,draft_generation FROM projects WHERE id=?1 AND user_id=?2", + rusqlite::params![&project_id, &auth.id], |r| Ok((r.get(0)?, r.get(1)?))) + .map_err(|_| AppError::not_found("project not found"))?; + if let Some(bh) = &base_hash { + if sha256_hex(draft.as_bytes()) != *bh { + return Err(AppError::conflict("stale_base", "current draft has changed since the proposal base; cannot auto-apply")); + } + } + if gen != base_gen { + return Err(AppError::conflict("stale_base", "proposal base does not match current generation")); + } + if gen != expected { + return Err(AppError::conflict("stale_generation", "draft changed; reload first")); + } + let hash = sha256_hex(proposed.as_bytes()); + let ts = now_iso(); + let vid = new_id(); + db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4", + rusqlite::params![&proposed, gen + 1, &ts, &project_id])?; + db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,'accepted AI proposal','ai',?5)", + rusqlite::params![&vid, &project_id, &proposed, &hash, &ts])?; + db.execute("UPDATE ai_requests SET status='succeeded' WHERE id=?1", [&aid]).ok(); + let mut st = db.prepare("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1")?; + let mut rows = st.query([&project_id])?; + let row = rows.next()?.ok_or_else(|| AppError::internal("project vanished"))?; + Ok(json!({ + "id": row.get::<_, String>(0)?, + "name": row.get::<_, String>(1)?, + "description": row.get::<_, String>(2)?, + "draft_code": row.get::<_, String>(3)?, + "draft_generation": row.get::<_, i64>(4)?, + "created_at": row.get::<_, String>(5)?, + "updated_at": row.get::<_, String>(6)?, + })) + }).await?; + audit(&cx, Some(&auth.id), "ai_accept", &aid, "ok").await; + Ok(Json(p)) +} + + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_takes_first_fenced_full_python_block() { + let s = "prose\n```python\nclass Strategy(bt.Strategy):\n pass\n```\ntrail"; + assert_eq!(extract_code_block(s).unwrap(), "class Strategy(bt.Strategy):\n pass\n"); + assert!(extract_code_block("no fence at all").is_none(), "no hardcoded fake suggestion"); + assert!(extract_code_block("```\n\n```").is_none(), "empty block rejected"); + } + + #[test] + fn usage_totals_never_invent_costs() { + // usage ledger records tokens measured, never price/cost + let j = json!({"usage": {"input_tokens": 11, "output_tokens": 7}}); + assert!(j.get("usage").map(|_| true).unwrap_or(false)); + assert!(!serde_json::to_string(&j["usage"]).unwrap().contains("cost")); + } + + #[test] + fn ai_session_headers_are_stable_app_identity_without_secrets() { + // Honest app UA: claims only this internal POC coding-assistant identity. + let ua = "strategy-lab-coding-assistant/0.1 (internal POC, strategy coding help only)"; + assert!(ua.starts_with("strategy-lab-coding-assistant/0.1")); + assert!(!ua.contains("opencode") && !ua.contains("curl"), "must not impersonate another client identity"); + // Stable per owner+project: same scope -> same id, different scope -> different id. + let owner = "6f1d2b3a-1111-4aaa-9bbb-cccccccccccc"; + let p1 = "00000000-2222-4333-8444-555555555555"; + let p2 = "00000000-2222-4333-8444-666666666666"; + let s1 = ai_session_id(owner, p1); + assert_eq!(s1, ai_session_id(owner, p1), "session must be stable across requests for owner+project"); + assert_ne!(s1, ai_session_id(owner, p2), "session is scoped to the project"); + assert_ne!(s1, ai_session_id("another-owner", p1), "session is scoped to the owner id"); + // No secret material ever rides in the session header. + assert!(!s1.contains(owner) && !s1.contains(p1)); + assert_eq!(s1.len(), 16 + 64, "prefix + sha256 hex of owner:project"); + } +} diff --git a/server/src/auth.rs b/server/src/auth.rs new file mode 100644 index 0000000..8942a83 --- /dev/null +++ b/server/src/auth.rs @@ -0,0 +1,713 @@ +use std::sync::Arc; +use axum::{ + extract::{FromRequestParts, Path}, + http::{header, request::Parts, HeaderMap, HeaderValue, StatusCode}, + response::{IntoResponse, Response}, + Json, +}; +use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier, password_hash::SaltString}; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +use crate::error::{AppError, AppResult}; +use crate::state::{AppState, Cx}; +use crate::util::{gen_token, new_id, now_iso, plus_hours, sha256_hex}; + +pub const COOKIE_NAME: &str = "sl_session"; + +const USER_COLS_FULL: &str = "id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at"; + +struct RowUser { + id: String, email: String, password_hash: String, name: String, + role: String, active: i64, ai_enabled: i64, daily_run_limit: i64, created_at: String, +} + +fn row_user(r: &rusqlite::Row) -> rusqlite::Result<RowUser> { + Ok(RowUser { + id: r.get(0)?, email: r.get(1)?, password_hash: r.get(2)?, name: r.get(3)?, + role: r.get(4)?, active: r.get(5)?, ai_enabled: r.get(6)?, daily_run_limit: r.get(7)?, created_at: r.get(8)?, + }) +} + +fn user_json(u: &RowUser) -> serde_json::Value { + json!({ + "id": u.id, "email": u.email, "name": u.name, "role": u.role, + "active": u.active != 0, "ai_enabled": u.ai_enabled != 0, + "daily_run_limit": u.daily_run_limit, "created_at": u.created_at, + }) +} + +#[derive(Debug, Clone, Serialize)] +pub struct User { + pub id: String, + pub email: String, + pub name: String, + pub role: String, + pub active: bool, + pub ai_enabled: bool, + pub daily_run_limit: i64, + pub created_at: String, +} + +#[derive(Clone)] +pub struct AuthUser { + pub id: String, + pub email: String, + pub role: String, + pub ai_enabled: bool, + pub daily_run_limit: i64, + pub session_id: String, +} + +impl FromRequestParts<Arc<AppState>> for AuthUser { + type Rejection = AppError; + async fn from_request_parts(parts: &mut Parts, state: &Arc<AppState>) -> Result<Self, Self::Rejection> { + let Some(token) = cookie_token(&parts.headers) else { + return Err(AppError::unauthorized("authentication required")); + }; + if token.len() < 32 { + return Err(AppError::unauthorized("invalid or expired session")); + } + let id = sha256_hex(token.as_bytes()); + let now = now_iso(); + let user = state.with_db(|db| { + db.query_row( + "SELECT u.id,u.email,u.role,u.ai_enabled,u.daily_run_limit,s.id FROM sessions s JOIN users u ON u.id=s.user_id WHERE s.id=?1 AND s.expires_at > ?2 AND u.active=1", + [&id, &now], + |r| Ok(AuthUser { id: r.get(0)?, email: r.get(1)?, role: r.get(2)?, ai_enabled: r.get::<_, i64>(3)? != 0, daily_run_limit: r.get(4)?, session_id: r.get(5)? }), + ).ok() + }).await; + user.ok_or_else(|| AppError::unauthorized("invalid or expired session")) + } +} + +/// Argon2id password hashing with a fresh random salt per call. +pub fn hash_password(p: &str) -> AppResult<String> { + let salt = SaltString::generate(&mut rand::rngs::OsRng); + Argon2::default().hash_password(p.as_bytes(), &salt) + .map(|h| h.to_string()) + .map_err(|e| AppError::internal(format!("hash: {e}"))) +} + +pub fn verify_password(hash: &str, p: &str) -> bool { + PasswordHash::new(hash).ok() + .and_then(|h| Argon2::default().verify_password(p.as_bytes(), &h).ok()) + .is_some() +} + +pub async fn fetch_user(cx: &Cx, id: &str) -> AppResult<User> { + cx.with_db(|db| { + db.query_row( + "SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users WHERE id=?1", + [id], + |r| Ok(User { id: r.get(0)?, email: r.get(1)?, name: r.get(2)?, role: r.get(3)?, active: r.get::<_, i64>(4)? != 0, ai_enabled: r.get::<_, i64>(5)? != 0, daily_run_limit: r.get(6)?, created_at: r.get(7)? }), + ).map_err(|_| AppError::not_found("user not found")) + }).await +} + +pub fn cookie_token(headers: &HeaderMap) -> Option<String> { + let raw = headers.get(header::COOKIE)?.to_str().ok()?; + raw.split(';').find_map(|c| { + let c = c.trim(); + c.strip_prefix(COOKIE_NAME) + .and_then(|v| v.strip_prefix('=')) + .map(str::to_string) + .filter(|v| !v.is_empty()) + }) +} + +/// Issue a random 256-bit session token; only sha256(token) is stored. +pub async fn session_cookie(cx: &Cx, user_id: &str) -> AppResult<(String, String)> { + let token = gen_token(); + let id = sha256_hex(token.as_bytes()); + let expires = plus_hours(cx.cfg.session_hours); + cx.with_db(|db| -> AppResult<()> { + db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES (?1,?2,?3,?4)", + rusqlite::params![&id, user_id, now_iso(), &expires])?; + Ok(()) + }).await?; + let mut cookie = format!("{COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age={}", cx.cfg.session_hours * 3600); + if cx.cfg.secure_cookies { + cookie.push_str("; Secure"); + } + Ok((cookie, expires)) +} + +pub fn clear_cookie() -> String { + format!("{COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0") +} + +pub async fn revoke_all_sessions(cx: &Cx, user_id: &str) -> AppResult<()> { + cx.with_db(|db| -> AppResult<()> { + db.execute("DELETE FROM sessions WHERE user_id=?1", [user_id])?; + Ok(()) + }).await +} + +/// Write a sanitized audit event. Callers must never pass passwords, keys, +/// tokens, strategy source or object payloads; only fixed action labels and +/// id/email targets, which sanitize_target bounds before storage. +pub async fn audit(cx: &Cx, actor: Option<&str>, action: &str, target: &str, status: &str) { + let target = sanitize_target(target); + cx.with_db(|db| { + db.execute("INSERT INTO audit (ts,actor_id,action,target,status) VALUES (?1,?2,?3,?4,?5)", + rusqlite::params![now_iso(), actor.map(str::to_string), action, target, status]).ok() + }).await; +} + +fn sanitize_target(t: &str) -> String { + if t.len() > 300 || t.lines().count() > 3 { + "redacted-oversized".to_string() + } else { + t.to_string() + } +} + +fn with_cookie(mut resp: Response, cookie: &str) -> Response { + if let Ok(hv) = HeaderValue::from_str(cookie) { + resp.headers_mut().insert(header::SET_COOKIE, hv); + } + resp +} + +async fn record_failure(cx: &Cx, email: &str) { + cx.with_db(|db| { + db.execute("INSERT INTO login_failures (email,failed_at) VALUES (?1,?2)", + rusqlite::params![email.to_string(), now_iso()]).ok() + }).await; +} + +/// Admin-issued one-time hashed token ("invitations" or "password_resets"). +/// Returns (raw_token shown once, expires_at). Only sha256 is persisted. +pub async fn make_admin_token( + cx: &Cx, + table: &str, + user_id: Option<&str>, + hours: i64, + email: Option<&str>, +) -> AppResult<(String, String)> { + let token = gen_token(); + let th = sha256_hex(token.as_bytes()); + let expires = plus_hours(hours); + let id = new_id(); + let email = email.map(|e| e.trim().to_lowercase()); + cx.with_db(|db| -> AppResult<()> { + match table { + "invitations" => { + db.execute("INSERT INTO invitations (id,email,token_hash,role,expires_at,created_at) VALUES (?1,?2,?3,'member',?4,?5)", + rusqlite::params![&id, email, &th, &expires, now_iso()])?; + } + "password_resets" => { + let uid = user_id.ok_or_else(|| AppError::bad("validation", "reset requires target user"))?; + db.execute("INSERT INTO password_resets (id,user_id,token_hash,expires_at) VALUES (?1,?2,?3,?4)", + rusqlite::params![&id, uid, &th, &expires])?; + } + other => return Err(AppError::internal(format!("unknown token table {other}"))), + } + Ok(()) + }).await?; + Ok((token, expires)) +} + +pub async fn purge_expired(cx: &Cx) { + cx.with_db(|db| { + let now = now_iso(); + db.execute("DELETE FROM sessions WHERE expires_at <= ?1", [&now]).ok(); + db.execute("DELETE FROM invitations WHERE expires_at <= ?1", [&now]).ok(); + db.execute("DELETE FROM password_resets WHERE expires_at <= ?1", [&now]).ok(); + db.execute("DELETE FROM login_failures WHERE failed_at <= ?1", [&plus_hours(-2)]).ok(); + }).await; +} + +// ---- handlers ---- + +#[derive(Deserialize)] +pub struct LoginReq { pub email: String, pub password: String } + +pub async fn login(cx: Cx, body: Option<Json<LoginReq>>) -> AppResult<Response> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let email = r.email.trim().to_lowercase(); + if email.is_empty() || r.password.is_empty() { + return Err(AppError::bad("validation", "email and password required")); + } + let failures: i64 = cx.with_db(|db| { + db.query_row("SELECT COUNT(*) FROM login_failures WHERE email=?1 AND failed_at >= ?2", + [&email, &plus_hours(-1)], |row| row.get(0)).unwrap_or(0) + }).await; + if failures > 20 { + audit(&cx, None, "login_throttled", &email, "rate_limited").await; + return Err(AppError::new(StatusCode::TOO_MANY_REQUESTS, "rate_limited", "too many failed logins; try again later")); + } + let Some(u) = cx.with_db(|db| { + db.query_row(&format!("SELECT {USER_COLS_FULL} FROM users WHERE email=?1"), [&email], row_user).ok() + }).await else { + record_failure(&cx, &email).await; + audit(&cx, None, "login_failed", &email, "invalid_credentials").await; + return Err(AppError::unauthorized("invalid credentials")); + }; + if !verify_password(&u.password_hash, &r.password) { + record_failure(&cx, &email).await; + audit(&cx, Some(&u.id), "login_failed", &u.id, "invalid_credentials").await; + return Err(AppError::unauthorized("invalid credentials")); + } + if u.active == 0 { + audit(&cx, Some(&u.id), "login_denied", &u.email, "disabled").await; + return Err(AppError::forbidden("account is disabled")); + } + purge_expired(&cx).await; + let (cookie, _) = session_cookie(&cx, &u.id).await?; + audit(&cx, Some(&u.id), "login", &u.id, "ok").await; + let resp = Json(user_json(&u)).into_response(); + Ok(with_cookie(resp, &cookie)) +} + +/// Registration payload has NO role field: role comes only from the +/// admin-issued invitation ("member" only in this POC). serde ignores extra +/// client fields, so role escalation at registration is impossible. +#[derive(Deserialize)] +pub struct RegisterReq { + pub invite_token: String, + pub name: String, + pub email: String, + pub password: String, +} + +pub async fn register(cx: Cx, body: Option<Json<RegisterReq>>) -> AppResult<Response> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if r.name.trim().is_empty() || r.name.trim().len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); } + if r.password.len() < 8 || r.password.len() > 256 { return Err(AppError::bad("validation", "password must be 8-256 characters")); } + let email = r.email.trim().to_lowercase(); + if !email.contains('@') || email.len() < 5 || email.len() > 320 { return Err(AppError::bad("validation", "valid email required")); } + let pw = hash_password(&r.password)?; + let token_hash = sha256_hex(r.invite_token.as_bytes()); + let day_limit = cx.cfg.default_run_limit_per_day; + let user_id: String = cx.with_db(|db| -> AppResult<String> { + // One-time atomically consumed invitation with email binding: the + // select, binding/role/expiry/used checks, user insert and invite + // marking all run inside ONE transaction; the marking ( + // UPDATE ... WHERE used_by IS NULL) guarantees concurrent retries lose. + let now = now_iso(); + let tx = db.transaction()?; + let invite: Option<(String, String, Option<String>)> = tx.query_row( + "SELECT id,role,email FROM invitations WHERE token_hash=?1", + [&token_hash], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).ok(); + let Some((inv_id, role, inv_email)) = invite else { + return Err(AppError::bad("invite_invalid", "invalid, used or expired invitation token")); + }; + if role != "member" { + return Err(AppError::bad("invite_invalid", "invitation does not permit this role")); + } + // Email binding: an invitation addressed to a specific email can only + // be consumed with that exact (case-insensitive) email. + if let Some(bound) = &inv_email { + let bound = bound.trim().to_lowercase(); + if !bound.is_empty() && bound != email { + return Err(AppError::bad("invite_email_mismatch", "invitation is bound to a different email")); + } + } + let expired = tx.query_row("SELECT expires_at FROM invitations WHERE id=?1", [&inv_id], |r| r.get::<_, String>(0)) + .map(|exp| exp <= now).unwrap_or(true); + if expired { + return Err(AppError::bad("invite_invalid", "invalid, used or expired invitation token")); + } + if tx.query_row("SELECT COUNT(*) FROM users WHERE email=?1", [&email], |c| c.get::<_, i64>(0)).unwrap_or(0) > 0 { + return Err(AppError::conflict("email_taken", "an account with this email already exists")); + } + let uid = new_id(); + tx.execute( + "INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,?4,'member',1,0,?5,?6)", + rusqlite::params![&uid, &email, &pw, r.name.trim(), day_limit, &now])?; + let n = tx.execute("UPDATE invitations SET used_by=?1 WHERE id=?2 AND used_by IS NULL AND expires_at > ?3", + rusqlite::params![&uid, &inv_id, &now])?; + if n != 1 { + return Err(AppError::conflict("invite_used", "invitation token already consumed")); + } + tx.commit()?; + Ok(uid) + }).await?; + audit(&cx, Some(&user_id), "register", &email, "ok").await; + purge_expired(&cx).await; + let (cookie, _) = session_cookie(&cx, &user_id).await?; + let u = fetch_user(&cx, &user_id).await?; + let resp = (StatusCode::CREATED, Json(json!({ "user": u }))).into_response(); + Ok(with_cookie(resp, &cookie)) +} + +pub async fn logout(cx: Cx, auth: AuthUser, headers: HeaderMap) -> AppResult<Response> { + if let Some(t) = cookie_token(&headers) { + let id = sha256_hex(t.as_bytes()); + cx.with_db(|db| { db.execute("DELETE FROM sessions WHERE id=?1", [&id]).ok() }).await; + } + audit(&cx, Some(&auth.id), "logout", &auth.id, "ok").await; + let resp = StatusCode::NO_CONTENT.into_response(); + Ok(with_cookie(resp, &clear_cookie())) +} + +pub async fn me(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> { + let u = fetch_user(&cx, &auth.id).await?; + Ok(Json(json!({ "user": u }))) +} + +#[derive(Deserialize)] +pub struct ProfileReq { pub name: String } + +pub async fn patch_profile(cx: Cx, auth: AuthUser, body: Option<Json<ProfileReq>>) -> AppResult<Json<serde_json::Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let name = r.name.trim().to_string(); + if name.is_empty() || name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); } + cx.with_db(|db| -> AppResult<()> { + let n = db.execute("UPDATE users SET name=?1 WHERE id=?2", rusqlite::params![&name, &auth.id])?; + if n == 0 { return Err(AppError::not_found("user not found")); } + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "profile_update", &auth.id, "ok").await; + let u = fetch_user(&cx, &auth.id).await?; + Ok(Json(json!({ "user": u }))) +} + +#[derive(Deserialize)] +pub struct PasswordReq { pub current_password: String, pub new_password: String } + +pub async fn change_password(cx: Cx, auth: AuthUser, headers: HeaderMap, body: Option<Json<PasswordReq>>) -> AppResult<Json<serde_json::Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if r.new_password.len() < 8 || r.new_password.len() > 256 { + return Err(AppError::bad("validation", "new password must be 8-256 characters")); + } + let hash: String = cx.with_db(|db| { + db.query_row("SELECT password_hash FROM users WHERE id=?1", [&auth.id], |row| row.get(0)) + .map_err(|_| AppError::unauthorized("authentication required")) + }).await?; + if !verify_password(&hash, &r.current_password) { + audit(&cx, Some(&auth.id), "password_change", &auth.id, "wrong_current_password").await; + return Err(AppError::unauthorized("current password incorrect")); + } + let new_hash = hash_password(&r.new_password)?; + // Keep the current session, revoke all others; update + revoke atomically. + let keep = cookie_token(&headers).map(|t| sha256_hex(t.as_bytes())).unwrap_or_default(); + cx.with_db(|db| -> AppResult<()> { + let tx = db.transaction()?; + tx.execute("UPDATE users SET password_hash=?1 WHERE id=?2", rusqlite::params![&new_hash, &auth.id])?; + tx.execute("DELETE FROM sessions WHERE user_id=?1 AND id != ?2", rusqlite::params![&auth.id, &keep])?; + tx.commit()?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "password_change", &auth.id, "ok").await; + Ok(Json(json!({ "ok": true, "other_sessions_revoked": true }))) +} + +/// Own-session listing, sanitized: sha256-derived ids only, never raw tokens; +/// expired sessions are never listed. `current` marks the caller's live session. +#[derive(Serialize)] +pub struct SessionInfo { pub id: String, pub created_at: String, pub expires_at: String } + +pub async fn list_sessions(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> { + let current = auth.session_id.clone(); + let rows: Vec<(String, String, String)> = cx.with_db(|db| { + let now = now_iso(); + let mut st = db.prepare("SELECT id,created_at,expires_at FROM sessions WHERE user_id=?1 AND expires_at > ?2 ORDER BY created_at DESC")?; + let rows = st + .query_map(rusqlite::params![&auth.id, &now], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))? + .collect::<Result<Vec<_>, _>>()?; + Ok(rows) + }).await.map_err(|e: rusqlite::Error| AppError::internal(e.to_string()))?; + Ok(Json(json!({ + "items": rows.into_iter().map(|(id, c, e)| json!({ + "id": id, "created_at": c, "expires_at": e, + "current": id == current, + })).collect::<Vec<_>>() + }))) +} + +pub async fn delete_session(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<serde_json::Value>> { + cx.with_db(|db| -> AppResult<()> { + if db.execute("DELETE FROM sessions WHERE id=?1 AND user_id=?2", rusqlite::params![&id, &auth.id])? == 0 { + // Never reveal another user's session ids. + return Err(AppError::not_found("session not found")); + } + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "session_revoke", &auth.id, "ok").await; + Ok(Json(json!({"ok": true}))) +} + +#[derive(Deserialize)] +pub struct ResetReq { pub token: String, pub new_password: String } + +pub async fn reset_password(cx: Cx, body: Option<Json<ResetReq>>) -> AppResult<Json<serde_json::Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if r.new_password.len() < 8 || r.new_password.len() > 256 { + return Err(AppError::bad("validation", "password must be 8-256 characters")); + } + let new_hash = hash_password(&r.new_password)?; + let th = sha256_hex(r.token.as_bytes()); + // Single-use consumption plus password update plus session revocation in + // ONE transaction: a second submission of the same token cannot succeed. + let uid: String = cx.with_db(|db| -> AppResult<String> { + let tx = db.transaction()?; + let (uid, used): (String, i64) = tx.query_row( + "SELECT user_id,used FROM password_resets WHERE token_hash=?1", + [&th], |row| Ok((row.get(0)?, row.get(1)?))) + .map_err(|_| AppError::bad("reset_invalid", "invalid or expired reset token"))?; + if used != 0 { + return Err(AppError::conflict("reset_used", "reset token already consumed")); + } + let expired = tx.query_row("SELECT expires_at FROM password_resets WHERE token_hash=?1", [&th], |row| row.get::<_, String>(0)) + .map(|exp| exp <= now_iso()).unwrap_or(true); + if expired { + return Err(AppError::bad("reset_invalid", "invalid or expired reset token")); + } + tx.execute("UPDATE password_resets SET used=1 WHERE token_hash=?1", [&th])?; + tx.execute("UPDATE users SET password_hash=?1 WHERE id=?2", rusqlite::params![&new_hash, &uid])?; + tx.execute("DELETE FROM sessions WHERE user_id=?1", [&uid])?; + tx.commit()?; + Ok(uid) + }).await?; + audit(&cx, Some(&uid), "password_reset", &uid, "ok").await; + Ok(Json(json!({"ok": true}))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::init_db; + + fn st() -> Arc<AppState> { + let conn = rusqlite::Connection::open_in_memory().expect("in-memory db"); + init_db(&conn).expect("schema"); + let cfg = crate::config::Config { + bind_addr: "127.0.0.1:0".into(), + canonical_origin: String::new(), + secure_cookies: false, + db_path: ":memory:".into(), + frontend_dir: "frontend/dist".into(), + data_dir: std::env::temp_dir().to_string_lossy().into_owned(), + worker_image: "test".into(), + fetch_timeout_secs: 1, + backtest_timeout_secs: 1, + run_concurrency: 1, + fetch_concurrency: 1, + session_hours: 24, + bootstrap_admin_email: None, + bootstrap_admin_password: None, + ai_base_url: "http://127.0.0.1:9".into(), + ai_model: "test-model".into(), + ai_daily_request_cap: 1, + ai_input_token_cap: 1, + ai_output_token_cap: 1, + ai_enabled_poc: false, + default_run_limit_per_day: 10, + version: "test".into(), + }; + Arc::new(AppState { + cfg, + db: tokio::sync::Mutex::new(conn), + run_sem: Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)), + }) + } + + fn handle(s: &Arc<AppState>) -> Cx { axum::extract::State(s.clone()) } + + fn cookie_of(token: &str) -> HeaderMap { + let mut hm = HeaderMap::new(); + hm.insert(header::COOKIE, HeaderValue::from_str(&format!("{COOKIE_NAME}={token}")).unwrap()); + hm + } + + /// Register through the real handler, then resolve the session exactly the + /// way the AuthUser extractor does, to produce a test auth context. + async fn register_ok(s: &Arc<AppState>, email: &str, name: &str, password: &str) -> (AuthUser, HeaderMap) { + let (inv, _) = make_admin_token(&handle(s), "invitations", None, 24, Some(email)).await.unwrap(); + let body = RegisterReq { + invite_token: inv, + name: name.into(), + email: email.into(), + password: password.into(), + }; + let resp = register(handle(s), Some(Json(body))).await.expect("register ok"); + let cookie = resp.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().to_string(); + let tok = cookie.split(';').next().unwrap() + .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('=')) + .expect("cookie contains the raw session token").to_string(); + let now = now_iso(); + let au = s.with_db(|db| { + db.query_row( + "SELECT u.id,u.email,u.role,u.ai_enabled,u.daily_run_limit,s.id FROM sessions s JOIN users u ON u.id=s.user_id WHERE s.id=?1 AND s.expires_at > ?2", + [&sha256_hex(tok.as_bytes()), &now], + |r| Ok(AuthUser { id: r.get(0)?, email: r.get(1)?, role: r.get(2)?, ai_enabled: r.get::<_, i64>(3)? != 0, daily_run_limit: r.get(4)?, session_id: r.get(5)? }), + ).unwrap() + }).await; + (au, cookie_of(&tok)) + } + + #[tokio::test] + async fn password_hashes_are_argon2() { + let h = hash_password("correct horse battery").unwrap(); + assert!(h.starts_with("$argon2")); + assert!(verify_password(&h, "correct horse battery")); + assert!(!verify_password(&h, "wrong password")); + } + + #[tokio::test] + async fn register_creates_member_and_ignores_client_role() { + let s = st(); + let (auth, _) = register_ok(&s, "[email protected]", "Tester User", "a-reasonable-passphrase-1").await; + assert_eq!(auth.email, "[email protected]"); + assert_eq!(auth.role, "member", "role must never come from client input"); + assert!(!auth.ai_enabled, "ai stays off by default for members"); + // No password hash ever appears in user-facing rows' JSON contract: + // the users table keeps the hash, API payloads never include it. + let u = fetch_user(&handle(&s), &auth.id).await.unwrap(); + let payload = serde_json::to_string(&u).unwrap(); + assert!(!payload.contains("password_hash")); + assert!(!payload.contains("argon2")); + } + + #[tokio::test] + async fn invitation_is_single_use() { + let s = st(); + let (inv, _) = make_admin_token(&handle(&s), "invitations", None, 24, None).await.unwrap(); + let body = RegisterReq { + invite_token: inv, + name: "Once".into(), + email: "[email protected]".into(), + password: "a-reasonable-passphrase-1".into(), + }; + register(handle(&s), Some(Json(body))).await.expect("first use ok"); + let consumed: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM invitations WHERE used_by IS NOT NULL", [], |r| r.get(0)).unwrap()).await; + assert_eq!(consumed, 1); + let token_rows: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM invitations WHERE used_by IS NULL", [], |r| r.get(0)).unwrap()).await; + assert_eq!(token_rows, 0, "invitation must be one-time"); + } + + #[tokio::test] + async fn invitation_email_binding_enforced() { + let s = st(); + let (inv, _) = make_admin_token(&handle(&s), "invitations", None, 24, Some("[email protected]")).await.unwrap(); + let body = RegisterReq { + invite_token: inv, + name: "Bound".into(), + email: "[email protected]".into(), + password: "a-reasonable-passphrase-1".into(), + }; + let err = register(handle(&s), Some(Json(body))).await.err().expect("must reject mismatched email"); + assert_eq!(err.code, "invite_email_mismatch"); + // The token was NOT consumed by the failed attempt. + let unused: i64 = handle(&s).with_db(|db| { + db.query_row("SELECT COUNT(*) FROM invitations WHERE used_by IS NULL", [], |r| r.get(0)).unwrap() + }).await; + assert_eq!(unused, 1); + } + + #[tokio::test] + async fn reset_password_is_single_use_and_revokes_sessions() { + let s = st(); + let (auth, _) = register_ok(&s, "[email protected]", "Reset User", "original-passphrase-1").await; + handle(&s).with_db(|db| { + db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES ('dead-session',?1,?2,?3)", + rusqlite::params![&auth.id, now_iso(), plus_hours(1)]).unwrap(); + }).await; + let (token, _) = make_admin_token(&handle(&s), "password_resets", Some(auth.id.as_str()), 1, None).await.unwrap(); + reset_password(handle(&s), Some(Json(ResetReq { token: token.clone(), new_password: "new-passphrase-1".into() }))).await.unwrap(); + let sessions: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM sessions WHERE user_id=?1", [&auth.id], |r| r.get(0)).unwrap()).await; + assert_eq!(sessions, 0, "reset must revoke all sessions"); + let second = reset_password(handle(&s), Some(Json(ResetReq { token, new_password: "new-passphrase-2".into() }))).await; + assert!(second.is_err(), "reset token must be single-use"); + let hash: String = handle(&s).with_db(|db| db.query_row( + "SELECT password_hash FROM users WHERE id=?1", [&auth.id], |r| r.get(0)).unwrap()).await; + assert!(verify_password(&hash, "new-passphrase-1")); + assert!(!verify_password(&hash, "new-passphrase-2")); + } + + #[tokio::test] + async fn change_password_revokes_others_keeps_current() { + let s = st(); + let (auth, headers) = register_ok(&s, "[email protected]", "Change User", "original-passphrase-1").await; + let extra = gen_token(); + let extra_id = sha256_hex(extra.as_bytes()); + handle(&s).with_db(|db| { + db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES (?1,?2,?3,?4)", + rusqlite::params![&extra_id, &auth.id, now_iso(), plus_hours(1)]).unwrap(); + }).await; + let done = change_password(handle(&s), auth.clone(), headers.clone(), Some(Json(PasswordReq { + current_password: "original-passphrase-1".into(), + new_password: "brand-new-passphrase".into(), + }))).await; + done.unwrap_or_else(|e| panic!("change failed: {e}")); + let remaining: i64 = handle(&s).with_db(|db| { + let now = now_iso(); + db.query_row("SELECT COUNT(*) FROM sessions WHERE user_id=?1 AND expires_at > ?2", [&auth.id, &now], |r| r.get(0)).unwrap() + }).await; + assert_eq!(remaining, 1, "only the current session survives"); + let wrong = change_password(handle(&s), auth.clone(), headers, Some(Json(PasswordReq { + current_password: "wrong".into(), + new_password: "another-passphrase-1".into(), + }))).await; + assert!(wrong.is_err(), "wrong current password must be rejected"); + } + + #[tokio::test] + async fn audit_never_records_secrets() { + let s = st(); + let long = format!("{}password=secret-value", "x".repeat(400)); + audit(&handle(&s), None, "login_failed", &long, "fail").await; + let stored: String = handle(&s).with_db(|db| db.query_row( + "SELECT target FROM audit ORDER BY seq DESC LIMIT 1", [], |r| r.get(0)).unwrap()).await; + assert_eq!(stored, "redacted-oversized"); + let dirty: i64 = handle(&s).with_db(|db| { + db.query_row("SELECT COUNT(*) FROM audit WHERE target LIKE '%password=%'", [], |r| r.get(0)).unwrap() + }).await; + assert_eq!(dirty, 0, "secrets must never be written into the audit trail"); + } + + #[tokio::test] + async fn cookie_sessions_are_hashed_never_plaintext() { + let s = st(); + handle(&s).with_db(|db| { + db.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('some-user','[email protected]','h','n','member',1,0,10,?1)", + [now_iso()]).unwrap(); + }).await; + let (cookie, _) = session_cookie(&handle(&s), "some-user").await.unwrap(); + let tok = cookie.split(';').next().unwrap() + .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('=')).unwrap(); + let plaintext: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM sessions WHERE id=?1", [&tok], |r| r.get(0)).unwrap()).await; + assert_eq!(plaintext, 0, "raw token must never be stored"); + let hashed: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM sessions WHERE id=?1", [&sha256_hex(tok.as_bytes())], |r| r.get(0)).unwrap()).await; + assert_eq!(hashed, 1); + assert!(cookie.contains("HttpOnly") && cookie.contains("SameSite=Strict")); + } + + #[tokio::test] + async fn login_throttles_after_repeated_failures() { + let s = st(); + for _ in 0..25 { + record_failure(&handle(&s), "[email protected]").await; + } + let res = login(handle(&s), Some(Json(LoginReq { + email: "[email protected]".into(), + password: "whatever-passphrase-1".into(), + }))).await; + match res { + Err(e) => assert_eq!(e.status, StatusCode::TOO_MANY_REQUESTS), + Ok(_) => panic!("login must be throttled after sustained failures"), + } + } + + #[tokio::test] + async fn logout_invalidates_the_server_session() { + let s = st(); + let (auth, headers) = register_ok(&s, "[email protected]", "Logout User", "a-reasonable-passphrase-1").await; + let tok = headers.get(header::COOKIE).unwrap().to_str().unwrap() + .split(';').next().unwrap() + .strip_prefix(COOKIE_NAME).and_then(|c| c.strip_prefix('=')).unwrap().to_string(); + logout(handle(&s), auth, headers).await.unwrap(); + let left: i64 = handle(&s).with_db(|db| db.query_row( + "SELECT COUNT(*) FROM sessions WHERE id=?1", [&sha256_hex(tok.as_bytes())], |r| r.get(0)).unwrap()).await; + assert_eq!(left, 0, "logout must delete the server-side session row"); + } +} diff --git a/server/src/config.rs b/server/src/config.rs new file mode 100644 index 0000000..410cb09 --- /dev/null +++ b/server/src/config.rs @@ -0,0 +1,65 @@ +use std::env; + +#[derive(Clone, Debug)] +pub struct Config { + pub bind_addr: String, + pub canonical_origin: String, + pub secure_cookies: bool, + pub db_path: String, + pub data_dir: String, + pub frontend_dir: String, + pub worker_image: String, + pub fetch_timeout_secs: u64, + pub backtest_timeout_secs: u64, + pub run_concurrency: usize, + pub fetch_concurrency: usize, + pub session_hours: i64, + pub bootstrap_admin_email: Option<String>, + pub bootstrap_admin_password: Option<String>, + pub ai_base_url: String, + pub ai_model: String, + pub ai_daily_request_cap: i64, + pub ai_input_token_cap: i64, + pub ai_output_token_cap: i64, + pub ai_enabled_poc: bool, + pub default_run_limit_per_day: i64, + pub version: String, +} + +fn envs(k: &str, d: &str) -> String { + env::var(k).ok().filter(|v| !v.is_empty()).unwrap_or_else(|| d.to_string()) +} + +impl Config { + pub fn from_env() -> Self { + let canonical_origin = envs("ORIGIN", ""); + Config { + bind_addr: envs("BIND", "127.0.0.1:8787"), + canonical_origin: canonical_origin.clone(), + secure_cookies: canonical_origin.starts_with("https://"), + db_path: envs("DB_PATH", "server-data/strategy-lab.sqlite3"), + data_dir: envs("DATA_DIR", "server-data/data"), + frontend_dir: envs("FRONTEND_DIR", "frontend/dist"), + worker_image: envs("WORKER_IMAGE", "strategy-lab-worker:local"), + fetch_timeout_secs: envs("FETCH_TIMEOUT_SECS", "1800").parse().unwrap_or(1800), + backtest_timeout_secs: envs("BACKTEST_TIMEOUT_SECS", "1800").parse().unwrap_or(1800), + run_concurrency: envs("RUN_CONCURRENCY", "1").parse().unwrap_or(1), + fetch_concurrency: envs("FETCH_CONCURRENCY", "2").parse().unwrap_or(2), + session_hours: envs("SESSION_HOURS", "336").parse().unwrap_or(336), + bootstrap_admin_email: env::var("BOOTSTRAP_ADMIN_EMAIL").ok(), + bootstrap_admin_password: env::var("BOOTSTRAP_ADMIN_PASSWORD").ok(), + ai_base_url: envs("AI_BASE_URL", "https://opencode.ai/zen/go/v1"), + ai_model: envs("AI_MODEL", "glm-5.3-flash"), + ai_daily_request_cap: envs("AI_DAILY_REQUEST_CAP", "20").parse().unwrap_or(20), + ai_input_token_cap: envs("AI_INPUT_TOKEN_CAP", "8000").parse().unwrap_or(8000), + ai_output_token_cap: envs("AI_OUTPUT_TOKEN_CAP", "6000").parse().unwrap_or(6000), + ai_enabled_poc: envs("AI_ENABLED_POC", "true") == "true", + default_run_limit_per_day: envs("DEFAULT_RUN_LIMIT_PER_DAY", "10").parse().unwrap_or(10), + version: envs("APP_VERSION", env!("CARGO_PKG_VERSION")), + } + } +} + +pub const MAX_CODE_LEN: usize = 256_000; +pub const MAX_SYMBOLS: usize = 5; +pub const MAX_RANGE_YEARS: i64 = 15; diff --git a/server/src/datasets.rs b/server/src/datasets.rs new file mode 100644 index 0000000..5b2428f --- /dev/null +++ b/server/src/datasets.rs @@ -0,0 +1,244 @@ +use axum::{extract::{Path, State}, Json}; +use chrono::NaiveDate; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::auth::{audit, AuthUser}; +use crate::config::{MAX_RANGE_YEARS, MAX_SYMBOLS}; +use crate::error::{AppError, AppResult}; +use crate::state::S; +use crate::util::{canonical_json, new_id, now_iso, sha256_hex}; + +/// Matches the fixed backend contract `Cx`; wired by State extractor in main.rs. +pub type Cx = State<S>; +const FIELDS: [&str; 6] = ["open", "high", "low", "close", "volume", "adj_factor"]; + + +#[derive(Deserialize, Serialize, Clone)] +pub struct InstrumentReq { + pub symbol: String, + pub market: String, + pub asset_type: String, + pub name: Option<String>, +} + +#[derive(Deserialize)] +pub struct DatasetRequest { + /// Optional: SPEC/UI permit an absent or blank name; the backend then + /// auto-generates a descriptive name and persists it. + /// Optional: SPEC/UI permit an absent or blank name; the backend then + /// auto-generates a descriptive name and persists it. + #[serde(default)] + pub name: Option<String>, + pub instruments: Vec<InstrumentReq>, + pub start_date: String, + pub end_date: String, + pub frequency: String, + pub adjustment: String, + pub fields: Vec<String>, +} + +pub fn canonical_request_value(req: &DatasetRequest) -> Value { + json!({ + "instruments": req.instruments.iter().map(|i| json!({ + "symbol": i.symbol, "market": i.market, "asset_type": i.asset_type, + "name": i.name.clone(), + })).collect::<Vec<_>>(), + "start_date": req.start_date, "end_date": req.end_date, + "frequency": req.frequency, "adjustment": req.adjustment, "fields": req.fields, + }) +} + +pub fn validate_request(req: &DatasetRequest) -> AppResult<String> { + if req.instruments.is_empty() || req.instruments.len() > MAX_SYMBOLS { + return Err(AppError::bad("validation", format!("instruments must be 1-{} items", MAX_SYMBOLS))); + } + if req.frequency != "daily" { return Err(AppError::bad("validation", "only daily frequency is supported")); } + if !matches!(req.adjustment.as_str(), "none" | "qfq" | "hfq") { + return Err(AppError::bad("validation", "adjustment must be none|qfq|hfq")); + } + if req.fields.is_empty() { return Err(AppError::bad("validation", "fields must not be empty")); } + for f in &req.fields { + if !FIELDS.contains(&f.as_str()) { return Err(AppError::bad("validation", format!("unsupported field: {f}"))); } + } + let mut seen = std::collections::HashSet::new(); + for i in &req.instruments { + if i.symbol.trim().is_empty() || i.market.trim().is_empty() { + return Err(AppError::bad("validation", "each instrument needs a symbol and a market")); + } + if !matches!(i.asset_type.as_str(), "stock" | "etf" | "index") { + return Err(AppError::bad("validation", "asset_type must be stock|etf|index")); + } + if i.asset_type == "index" && req.adjustment != "none" { + return Err(AppError::bad("validation", "index instruments support adjustment 'none' only (explicit restriction, no factors)")); + } + if !seen.insert(format!("{}|{}|{}", i.market, i.asset_type, i.symbol)) { + return Err(AppError::bad("validation", "duplicate instrument in request")); + } + } + let sd = NaiveDate::parse_from_str(&req.start_date, "%Y-%m-%d").map_err(|_| AppError::bad("validation", "start_date must be YYYY-MM-DD"))?; + let ed = NaiveDate::parse_from_str(&req.end_date, "%Y-%m-%d").map_err(|_| AppError::bad("validation", "end_date must be YYYY-MM-DD"))?; + if ed < sd { return Err(AppError::bad("validation", "end_date must not precede start_date")); } + if (ed - sd).num_days() > MAX_RANGE_YEARS * 366 { + return Err(AppError::bad("validation", "range exceeds maximum of 15 years")); + } + Ok(sha256_hex(canonical_json(&canonical_request_value(req)).as_bytes())) +} + +/// Full stored manifest JSON -> client copy without host/internal paths. +pub fn client_manifest(m: &Value) -> Value { + let mut o = m.clone(); + o.as_object_mut().map(|m| m.remove("preview")); + if let Some(objs) = o.get_mut("objects").and_then(|v| v.as_array_mut()) { + for obj in objs.iter_mut() { + if let Some(map) = obj.as_object_mut() { + map.remove("path"); + } + } + } + o +} + +fn row_dataset(r: &rusqlite::Row) -> rusqlite::Result<Value> { + let manifest: Option<String> = r.get(6)?; + let manifest_v: Value = manifest.and_then(|m| serde_json::from_str::<Value>(&m).ok()).unwrap_or(Value::Null); + // The stored canonical request is persisted as a JSON string; clients get an object. + let request_s: String = r.get(2)?; + let request_v: Value = serde_json::from_str::<Value>(&request_s) + .map_err(|_| rusqlite::Error::InvalidColumnType(2, "dataset request".into(), rusqlite::types::Type::Text))?; + Ok(json!({ + "id": r.get::<_, String>(0)?, + "name": r.get::<_, String>(1)?, + "request": request_v, + "status": r.get::<_, String>(3)?, + "error": r.get::<_, Option<String>>(4)?, + "cache_hit": r.get::<_, Option<i64>>(5)?.map(|v| v != 0), + "manifest": if manifest_v.is_null() { Value::Null } else { client_manifest(&manifest_v) }, + "warnings": manifest_v.get("warnings").cloned().unwrap_or(json!([])), + "manifest_hash": manifest_v.get("hash").cloned().unwrap_or(Value::Null), + "created_at": r.get::<_, String>(7)?, + "updated_at": r.get::<_, String>(8)?, + })) +} + +pub async fn assert_owned(cx: &Cx, user_id: &str, dataset_id: &str) -> AppResult<()> { + let found: Option<String> = cx.with_db(|db| { + db.query_row("SELECT user_id FROM datasets WHERE id=?1", [dataset_id], |r| r.get(0)).ok() + }).await; + match found { + Some(o) if o == user_id => Ok(()), + _ => Err(AppError::not_found("dataset not found")), + } +} + +async fn query_dataset(cx: &Cx, sql: &str, dataset_id: &str) -> AppResult<Value> { + cx.with_db(move |db| { + db.query_row(sql, [dataset_id], row_dataset) + .map_err(|_| AppError::not_found("dataset not found")) + }).await +} + +const LIST_SQL: &str = "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE user_id=?1 ORDER BY created_at DESC LIMIT 200"; + +pub async fn list(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> { + let items: Vec<Value> = cx.with_db(|db| { + let mut st = db.prepare(LIST_SQL)?; + let mut rows = st.query([auth.id.clone()])?; + let mut out = Vec::new(); + while let Some(r) = rows.next()? { out.push(row_dataset(r)?); } + Ok::<_, AppError>(out) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + assert_owned(&cx, &auth.id, &id).await?; + let v = query_dataset(&cx, "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE id=?1", &id).await?; + Ok(Json(v)) +} + +/// Load the stored worker manifest (with internal paths) of a ready dataset. +pub async fn load_manifest(cx: &Cx, dataset_id: &str) -> AppResult<Value> { + let m: String = cx.with_db(|db| { + db.query_row("SELECT manifest FROM datasets WHERE id=?1 AND status='ready'", [dataset_id], |r| r.get(0)) + .map_err(|_| AppError::conflict("dataset_not_ready", "dataset not ready")) + }).await?; + serde_json::from_str(&m).map_err(|e| AppError::internal(format!("manifest corrupt: {e}"))) +} + +pub async fn create(cx: Cx, auth: AuthUser, body: Option<Json<DatasetRequest>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + // Frontend sends name: optional/empty; default is an auto-generated descriptive name. + let name = if r.name.as_deref().map(|n| n.trim().is_empty()).unwrap_or(true) { + // Frontend allows an empty name to auto-generate: instruments + range. + let syms: Vec<String> = r.instruments.iter().map(|i| i.symbol.clone()).collect(); + format!("{} · {} ~ {}", syms.join(","), r.start_date, r.end_date) + } else { + r.name.as_deref().unwrap_or_default().trim().to_string() + }; + if name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200)")); } + let _key = validate_request(&r)?; + let stored_request = canonical_request_value(&r); + let id = new_id(); + let ts = now_iso(); + let uid = auth.id.clone(); + cx.with_db(|db| -> AppResult<()> { + db.execute("INSERT INTO datasets (id,user_id,name,request,status,cache_hit,created_at,updated_at) VALUES (?1,?2,?3,?4,'pending',0,?5,?5)", + rusqlite::params![&id, &uid, &name, stored_request.to_string(), &ts])?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "dataset_create", &id, "ok").await; + let v = query_dataset(&cx, "SELECT id,name,request,status,error,cache_hit,manifest,created_at,updated_at FROM datasets WHERE id=?1", &id).await?; + Ok((axum::http::StatusCode::ACCEPTED, Json(v))) +} + +pub async fn preview(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + assert_owned(&cx, &auth.id, &id).await?; + let m = load_manifest(&cx, &id).await?; + let p = m.get("preview").cloned().unwrap_or(Value::Null); + if p.is_null() { return Err(AppError::not_found("preview not available yet")); } + Ok(Json(p)) +} + + +#[cfg(test)] +mod tests { + use super::*; + + fn req() -> DatasetRequest { + serde_json::from_value(json!({ + "name": "t", "instruments": [{"symbol": "600000", "market": "cn", "asset_type": "stock", "name": "浦发银行"}], + "start_date": "2024-01-01", "end_date": "2024-06-30", + "frequency": "daily", "adjustment": "none", "fields": ["open","high","low","close","volume"] + })).unwrap() + } + + #[test] + fn validate_rejects_unsupported() { + assert_eq!(validate_request(&req()).unwrap().len(), 64); + let mut r = req(); r.frequency = "hourly".into(); + assert_eq!(validate_request(&r).unwrap_err().code, "validation"); + let mut r = req(); r.adjustment = "qfq".into(); r.instruments[0].asset_type = "index".into(); + assert_eq!(validate_request(&r).unwrap_err().code, "validation", "index+adjustment must be explicit rejections"); + let mut r = req(); r.instruments.push(r.instruments[0].clone()); + assert_eq!(validate_request(&r).unwrap_err().code, "validation", "duplicate instruments rejected"); + } + + #[test] + fn cache_key_is_stable_and_shared_regardless_of_display_name() { + let other = { let mut o = req(); o.name = Some("别的名字".to_string()); o }; + assert_eq!(canonical_request_value(&req()), canonical_request_value(&other)); + assert_eq!(validate_request(&req()).unwrap(), validate_request(&other).unwrap()); + } + + #[test] + fn client_manifest_strips_internal_paths_and_preview() { + let m = json!({ + "hash": "h", "warnings": [], "preview": {"rows": [1]}, + "objects": [{"instrument": {"symbol": "SH#600000"}, "path": "objects/ab/ab12.csv"}] + }); + let c = client_manifest(&m); + assert!(serde_json::to_string(&c).unwrap().find("objects/ab").is_none(), "host paths must not leak"); + assert_eq!(c.get("hash"), Some(&json!("h"))); + } +} diff --git a/server/src/db.rs b/server/src/db.rs new file mode 100644 index 0000000..d8b09dd --- /dev/null +++ b/server/src/db.rs @@ -0,0 +1,146 @@ +use rusqlite::Connection; + +pub fn init_db(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch( + r#" + PRAGMA journal_mode=WAL; + PRAGMA foreign_keys=ON; + CREATE TABLE IF NOT EXISTS users ( + id TEXT PRIMARY KEY, + email TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL, + name TEXT NOT NULL, + role TEXT NOT NULL CHECK(role IN ('admin','member')), + active INTEGER NOT NULL DEFAULT 1, + ai_enabled INTEGER NOT NULL DEFAULT 0, + daily_run_limit INTEGER NOT NULL DEFAULT 10, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, -- sha256(token) + user_id TEXT NOT NULL REFERENCES users(id), + created_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + origin_note TEXT + ); + CREATE TABLE IF NOT EXISTS login_failures ( + email TEXT NOT NULL, + failed_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS invitations ( + id TEXT PRIMARY KEY, + email TEXT, + token_hash TEXT NOT NULL UNIQUE, + role TEXT NOT NULL DEFAULT 'member', + expires_at TEXT NOT NULL, + used_by TEXT, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS password_resets ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id), + token_hash TEXT NOT NULL UNIQUE, + expires_at TEXT NOT NULL, + used INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS audit ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + actor_id TEXT, + action TEXT NOT NULL, + target TEXT, + status TEXT NOT NULL, + details TEXT + ); + CREATE TABLE IF NOT EXISTS projects ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id), + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + draft_code TEXT NOT NULL DEFAULT '', + draft_generation INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS project_versions ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL REFERENCES projects(id), + code TEXT NOT NULL, + hash TEXT NOT NULL, + message TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('manual','run','ai','restore')), + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS datasets ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id), + name TEXT NOT NULL, + request TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ('pending','running','ready','failed')), + error TEXT, + manifest_hash TEXT, + manifest TEXT, + cache_hit INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS data_objects ( + hash TEXT PRIMARY KEY, + path TEXT NOT NULL UNIQUE, + size INTEGER NOT NULL, + fetched_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS runs ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id), + project_id TEXT NOT NULL REFERENCES projects(id), + version_id TEXT NOT NULL REFERENCES project_versions(id), + dataset_id TEXT NOT NULL REFERENCES datasets(id), + status TEXT NOT NULL CHECK(status IN ('queued','running','succeeded','failed','cancelled')), + config TEXT NOT NULL, + manifest_hash TEXT, + container_id TEXT, + result TEXT, + error TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT + ); + CREATE INDEX IF NOT EXISTS runs_project ON runs(project_id, created_at DESC); + CREATE INDEX IF NOT EXISTS runs_user_daily ON runs(user_id, created_at DESC); + CREATE TABLE IF NOT EXISTS ai_requests ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id), + project_id TEXT NOT NULL REFERENCES projects(id), + instruction TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ('pending','succeeded','failed')), + model TEXT, + explanation TEXT, + proposed_code TEXT, + diff TEXT, + base_generation INTEGER, + base_code_hash TEXT, + version_id TEXT, + usage TEXT, + error TEXT, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS fetch_cache ( + key TEXT PRIMARY KEY, + manifest_hash TEXT NOT NULL, + manifest TEXT NOT NULL, + object_count INTEGER NOT NULL, + fetched_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS ai_usage ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + request_id TEXT, + ts TEXT NOT NULL, + kind TEXT NOT NULL, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0 + ); + "#, + ) +} diff --git a/server/src/error.rs b/server/src/error.rs new file mode 100644 index 0000000..0342afd --- /dev/null +++ b/server/src/error.rs @@ -0,0 +1,145 @@ +use axum::{ + http::StatusCode, + response::{IntoResponse, Response}, + Json, +}; +use serde_json::{json, Value}; + +/// Coherent error type for all handlers and helpers. Exactly one set of +/// constructors, one IntoResponse, and the three owned From conversions +/// (rusqlite, std::io, serde_json) live here and nowhere else. +#[derive(Debug)] +pub struct AppError { + pub status: StatusCode, + pub code: &'static str, + pub message: String, + pub details: Option<Value>, +} + +impl std::fmt::Display for AppError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}: {}", self.code, self.message) + } +} + +impl std::error::Error for AppError {} + +impl AppError { + pub fn new(status: StatusCode, code: &'static str, message: impl Into<String>) -> Self { + AppError { status, code, message: message.into(), details: None } + } + pub fn bad(code: &'static str, msg: impl Into<String>) -> Self { + Self::new(StatusCode::BAD_REQUEST, code, msg) + } + pub fn unauthorized(msg: impl Into<String>) -> Self { + Self::new(StatusCode::UNAUTHORIZED, "unauthorized", msg) + } + pub fn forbidden(msg: impl Into<String>) -> Self { + Self::new(StatusCode::FORBIDDEN, "forbidden", msg) + } + pub fn not_found(msg: impl Into<String>) -> Self { + Self::new(StatusCode::NOT_FOUND, "not_found", msg) + } + pub fn conflict(code: &'static str, msg: impl Into<String>) -> Self { + Self::new(StatusCode::CONFLICT, code, msg) + } + pub fn internal(msg: impl Into<String>) -> Self { + Self::new(StatusCode::INTERNAL_SERVER_ERROR, "internal", msg) + } + pub fn with_details(mut self, d: Value) -> Self { + self.details = Some(d); + self + } + pub fn with_code(mut self, code: &'static str) -> Self { + self.code = code; + self + } +} + +impl IntoResponse for AppError { + fn into_response(self) -> Response { + let mut e = json!({"code": self.code, "message": self.message}); + if let Some(details) = self.details { + e["details"] = details; + } + (self.status, Json(json!({"error": e}))).into_response() + } +} + +impl From<rusqlite::Error> for AppError { + fn from(e: rusqlite::Error) -> Self { + match &e { + rusqlite::Error::QueryReturnedNoRows => AppError::not_found("resource not found"), + _ => { + tracing::error!("db error: {e}"); + AppError::internal("internal storage error") + } + } + } +} + +impl From<std::io::Error> for AppError { + fn from(e: std::io::Error) -> Self { + tracing::error!("io error: {e}"); + AppError::internal("internal filesystem error") + } +} + +impl From<serde_json::Error> for AppError { + fn from(e: serde_json::Error) -> Self { + tracing::error!("json error: {e}"); + AppError::internal("internal json error") + } +} + +pub type AppResult<T> = Result<T, AppError>; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn constructors_map_to_expected_status_and_code() { + let e = AppError::bad("validation", "x"); + assert_eq!(e.status, StatusCode::BAD_REQUEST); + assert_eq!(e.code, "validation"); + assert_eq!(AppError::unauthorized("y").status, StatusCode::UNAUTHORIZED); + assert_eq!(AppError::forbidden("y").status, StatusCode::FORBIDDEN); + assert_eq!(AppError::not_found("y").status, StatusCode::NOT_FOUND); + assert_eq!(AppError::conflict("stale", "y").status, StatusCode::CONFLICT); + assert_eq!(AppError::internal("y").status, StatusCode::INTERNAL_SERVER_ERROR); + let v = AppError::conflict("stale", "y") + .with_code("dataset_not_ready") + .with_details(json!({"warnings": [1]})); + assert_eq!(v.code, "dataset_not_ready"); + assert_eq!(v.details.expect("details")["warnings"], json!([1])); + } + + #[tokio::test] + async fn body_shape_is_error_code_message_details() { + let mut resp = AppError::conflict("stale", "expected_generation out of date") + .with_details(json!({"expected": 3})) + .into_response(); + let bytes = axum::body::to_bytes(std::mem::take(resp.body_mut()), 64 * 1024).await.unwrap(); + let parsed: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(parsed["error"]["code"], "stale"); + assert_eq!(parsed["error"]["message"], "expected_generation out of date"); + assert_eq!(parsed["error"]["details"]["expected"], 3); + + let mut plain = AppError::internal("boom").into_response(); + let plain: Value = serde_json::from_slice( + &axum::body::to_bytes(std::mem::take(plain.body_mut()), 64 * 1024).await.unwrap(), + ).unwrap(); + assert!(plain["error"].get("details").is_none()); + } + + #[test] + fn from_impls_convert_without_duplicates() { + let e: AppError = rusqlite::Error::QueryReturnedNoRows.into(); + assert_eq!(e.status, StatusCode::NOT_FOUND); + let e: AppError = std::io::Error::new(std::io::ErrorKind::NotFound, "nope").into(); + assert_eq!(e.status, StatusCode::INTERNAL_SERVER_ERROR); + let e: AppError = serde_json::from_str::<Value>("{").unwrap_err().into(); + assert_eq!(e.status, StatusCode::INTERNAL_SERVER_ERROR); + } +} diff --git a/server/src/jobs.rs b/server/src/jobs.rs new file mode 100644 index 0000000..a3c49eb --- /dev/null +++ b/server/src/jobs.rs @@ -0,0 +1,857 @@ +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::Mutex; +use std::collections::HashSet; +use serde_json::{json, Value}; + +use crate::auth::audit; +use crate::error::AppError; +use crate::state::AppState; +use crate::store::ObjectStore; +use crate::util::{canonical_json, new_id, now_iso, sha256_hex}; + +pub struct Signals; + +impl Signals { + pub fn new() -> Self { Signals } +} + +pub async fn main_loop(cx: Arc<AppState>, _signals: Signals) { + loop { + if let Err(e) = tick(&cx).await { tracing::error!("job loop: {e}"); } + tokio::time::sleep(Duration::from_millis(700)).await; + } +} + +// ---- shared in-flight fetch serialization for the exact request cache key ---- + +fn inflight() -> &'static Mutex<HashSet<String>> { + static SET: std::sync::OnceLock<Mutex<HashSet<String>>> = std::sync::OnceLock::new(); + SET.get_or_init(|| Mutex::new(HashSet::new())) +} + +pub fn cache_key_for_request(req: &Value) -> String { + sha256_hex(canonical_json(req).as_bytes()) +} + +async fn release_key(key: &str) { + inflight().lock().await.remove(key); +} + +// ---- dispatcher ---- + +async fn tick(cx: &Arc<AppState>) -> Result<(), AppError> { + // pending datasets -> bounded fetch concurrency + let n_running: i64 = cx.with_db(|db| { + db.query_row("SELECT COUNT(*) FROM datasets WHERE status='running'", [], |r| r.get(0)) + }).await.unwrap_or(0); + if n_running < cx.cfg.fetch_concurrency as i64 { + let next: Option<(String, String)> = cx.with_db(|db| { + db.query_row("SELECT id,request FROM datasets WHERE status='pending' ORDER BY created_at LIMIT 1", [], + |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok() + }).await; + if let Some((did, request)) = next { + cx.with_db(|db| { + db.execute("UPDATE datasets SET status='running', updated_at=?1 WHERE id=?2 AND status='pending'", + rusqlite::params![now_iso(), &did]).ok(); + }).await; + let cx2 = cx.clone(); + tokio::spawn(async move { + if let Err(e) = spawn_fetch_job(&cx2, &did, &request).await { + tracing::error!("fetch {did}: {e}"); + cx2.with_db(|db| { + db.execute("UPDATE datasets SET status='failed', error=?1, updated_at=?2 WHERE id=?3 AND status IN ('pending','running')", + rusqlite::params![e.message, now_iso(), &did]).ok(); + }).await; + } + }); + } + } + + // queued runs -> single active backtest worker (POC) + let n_running_runs: i64 = cx.with_db(|db| { + db.query_row("SELECT COUNT(*) FROM runs WHERE status='running'", [], |r| r.get(0)) + }).await.unwrap_or(0); + if n_running_runs >= cx.cfg.run_concurrency as i64 { return Ok(()); } + let next: Option<(String, String)> = cx.with_db(|db| { + db.query_row("SELECT id,config FROM runs WHERE status='queued' ORDER BY created_at LIMIT 1", [], + |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok() + }).await; + let Some((run_id, config)) = next else { return Ok(()); }; + let container_name = format!("sl-run-{}", new_id()); + // Atomic claim: queued -> running WITH container_id set in the same transaction. + // A run never becomes 'running' without a container identity, so the claim + // itself proves the worker task was handed off. + let claimed = cx.with_db(|db| -> Result<bool, AppError> { + move_claim(&mut *db, &run_id, &container_name) + }).await?; + if !claimed { return Ok(()); } + let cx2 = cx.clone(); + tokio::spawn(async move { + let user_id: String = cx2.with_db(|db| { + db.query_row("SELECT user_id FROM runs WHERE id=?1", [&run_id], |r| r.get(0)).unwrap_or_default() + }).await; + let res = run_backtest(&cx2, &run_id, &config, &container_name).await; + let _ = finalize_run(&cx2, &run_id, &user_id, res).await; + }); + Ok(()) +} + +/// Recheck account/quota/dataset durability, then claim queued -> running with +/// a container id atomically inside one transaction. +/// Returns true only if this caller actually claimed the run (and must spawn the worker). +fn move_claim(db: &mut rusqlite::Connection, run_id: &str, container_id: &str) -> Result<bool, AppError> { + db.execute("BEGIN IMMEDIATE", []).ok(); + match db.query_row( + "SELECT u.active, r.user_id FROM runs r JOIN users u ON u.id=r.user_id WHERE r.id=?1", + [run_id], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?))) { + Ok((active, _uid)) if active == 1 => {}, + _ => { + db.execute("UPDATE runs SET status='failed', error='account disabled', finished_at=?1 WHERE id=?2 AND status='queued'", + rusqlite::params![now_iso(), run_id]).ok(); + db.execute("COMMIT", []).ok(); + return Ok(false); + } + } + let uid = db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get::<_, String>(0)).unwrap_or_default(); + let ready_ok: Option<i64> = db.query_row( + "SELECT d.status='ready' AS ok FROM datasets d JOIN runs r ON r.dataset_id=d.id AND r.user_id=d.user_id WHERE r.id=?1", + [run_id], |r| r.get::<_, i64>(0)).ok(); + if ready_ok != Some(1) { + db.execute("UPDATE runs SET status='failed', error='dataset no longer ready at dequeue', finished_at=?1 WHERE id=?2 AND status='queued'", + rusqlite::params![now_iso(), run_id]).ok(); + db.execute("COMMIT", []).ok(); + return Ok(false); + } + let today = format!("{}%", chrono::Utc::now().format("%Y-%m-%d").to_string()); + let used: i64 = db.query_row("SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2", + rusqlite::params![&uid, &today], |r| r.get(0)).unwrap_or(0); + let limit: i64 = db.query_row("SELECT daily_run_limit FROM users WHERE id=?1", [&uid], |r| r.get(0)).unwrap_or(10); + if used > limit { + db.execute("UPDATE runs SET status='failed', error='run quota exceeded', finished_at=?1 WHERE id=?2 AND status='queued'", + rusqlite::params![now_iso(), run_id]).ok(); + db.execute("COMMIT", []).ok(); + return Ok(false); + } + // Single atomic transition: a claimed run is running WITH its container identity. + let n = db.execute( + "UPDATE runs SET status='running', started_at=?1, container_id=?2 WHERE id=?3 AND status='queued'", + rusqlite::params![now_iso(), container_id, run_id]).unwrap_or(0); + db.execute("COMMIT", []).ok(); + Ok(n == 1) +} + +async fn finalize_run(cx: &Arc<AppState>, run_id: &str, user_id: &str, res: Result<Value, AppError>) { + cx.with_db(|db| -> Result<(), AppError> { + match res { + Ok(result) => { + let n = db.execute("UPDATE runs SET status='succeeded', result=?1, error=NULL, finished_at=?2 WHERE id=?3 AND status='running'", + rusqlite::params![result.to_string(), now_iso(), run_id]).unwrap_or(0); + if n > 0 { + let cxx = cx.clone(); + let uid = user_id.to_string(); + let rid = run_id.to_string(); + tokio::spawn(async move { audit(&axum::extract::State(cxx), Some(&uid), "run_succeeded", &rid, "ok").await }); + } + } + Err(e) => { + // guaranteed non-empty terminal reason, never an empty failure + let msg = if e.message.trim().is_empty() { "worker failed without error detail".to_string() } else { e.message.clone() }; + db.execute("UPDATE runs SET status='failed', error=?1, finished_at=?2 WHERE id=?3 AND status='running'", + rusqlite::params![msg, now_iso(), run_id]).ok(); + } + } + Ok(()) + }).await.ok(); +} + +// ---- dataset fetch ---- + +async fn try_cache(cx: &Arc<AppState>, dataset_id: &str, key: &str) -> Result<bool, AppError> { + let manifest: Option<String> = cx.with_db(|db| { + db.query_row("SELECT manifest FROM fetch_cache WHERE key=?1", [key], |r| r.get(0)).ok() + }).await; + let Some(m) = manifest else { return Ok(false); }; + let mv: Value = serde_json::from_str(&m).unwrap_or(Value::Null); + let hash = mv.get("hash").and_then(|v| v.as_str()).unwrap_or_default().to_string(); + cx.with_db(|db| { + db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=1, updated_at=?3 WHERE id=?4 AND status='running'", + rusqlite::params![&m, hash, now_iso(), dataset_id]).ok(); + }).await; + Ok(true) +} + +/// One dataset fetch: cache recheck or container run + artifact ingest. +pub async fn spawn_fetch_job(cx: &Arc<AppState>, dataset_id: &str, request_s: &str) -> Result<(), AppError> { + let request: Value = serde_json::from_str(request_s).map_err(|e| AppError::internal(format!("stored request invalid: {e}")))?; + let key = cache_key_for_request(&request); + + // serialize identical cache keys: second arrives late and rechecks cache + let deadline = std::time::Duration::from_secs(cx.cfg.fetch_timeout_secs.max(60)); + let started = std::time::Instant::now(); + loop { + { + let mut set = inflight().lock().await; + if set.insert(key.clone()) { break; } + + } + if try_cache(cx, dataset_id, &key).await? { return Ok(()); } + if started.elapsed() > deadline { + return Err(AppError::internal("waiting on identical in-flight fetch timed out")); + } + tokio::time::sleep(Duration::from_millis(1500)).await; + } + let result = perform_fetch(cx, dataset_id, request, &key).await; + release_key(&key).await; + result +} + +async fn perform_fetch(cx: &Arc<AppState>, dataset_id: &str, request: Value, key: &str) -> Result<(), AppError> { + // cached from a prior identical request? + if try_cache(cx, dataset_id, key).await? { return Ok(()); } + + let store = ObjectStore::new(&cx.cfg.data_dir); + let work = temp_job_dir(&cx.cfg.data_dir)?; + let input = work.join("input"); + let output = work.join("output"); + std::fs::create_dir_all(&input).map_err(io_err)?; + std::fs::create_dir_all(&output).map_err(io_err)?; + open_writable(&output)?; + + std::fs::write(input.join("request.json"), request.to_string()).map_err(io_err)?; + let mounts = vec![ + (input.display().to_string(), "/input".to_string(), true), + (output.display().to_string(), "/output".to_string(), false), + ]; + let args = vec![ + "python".into(), "-m".into(), "worker.main".into(), "fetch".into(), + "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(), + ]; + let name = format!("sl-fetch-{}", new_id()); + let res = crate::worker::run_named(&*cx, true, &mounts, &args, &name, cx.cfg.fetch_timeout_secs).await; + let out_take = match res { + Ok(r) if r.ok() => r, + Ok(r) => { + let msg = worker_error_message(&output, r.stderr.as_str()); + cleanup(&work); + return Err(AppError::bad("fetch_failed", msg)); + } + Err(e) => { cleanup(&work); return Err(e); } + }; + + let build = ingest_fetch_output(&store, &work).await; + let (manifest, _warnings) = match build { + Ok(v) => v, + Err(e) => { cleanup(&work); return Err(e); } + }; + let _ = out_take; + let manifest_hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string(); + if manifest_hash.is_empty() || manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) { + cleanup(&work); + return Err(AppError::internal("fetch produced no usable objects")); + } + let manifest_s = manifest.to_string(); + let object_count = manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.len()).unwrap_or(0) as i64; + cx.with_db(|db| -> Result<(), AppError> { + db.execute("INSERT OR REPLACE INTO fetch_cache (key,manifest_hash,manifest,object_count,fetched_at) VALUES (?1,?2,?3,?4,?5)", + rusqlite::params![&key, &manifest_hash, &manifest_s, object_count, now_iso()])?; + db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=0, updated_at=?3 WHERE id=?4 AND status='running'", + rusqlite::params![&manifest_s, &manifest_hash, now_iso(), dataset_id])?; + Ok(()) + }).await?; + cleanup(&work); + tracing::info!("fetch {dataset_id} cached under {key} ({object_count} objects)"); + Ok(()) +} + + +// manifest hash must be content identity: independent of user/request ids and fetch timestamp +fn manifest_content_hash(m: &Value) -> String { + let mut objects = Vec::new(); + for o in m.get("objects").and_then(|v| v.as_array()).unwrap_or(&vec![]).iter() { + let mut e = o.clone(); + if let Some(map) = e.as_object_mut() { + map.remove("fetched_at"); + map.remove("path"); + } + objects.push(e); + } + let payload = json!({ + "frequency": m.get("frequency"), + "adjustment": m.get("adjustment"), + "schema_version": m.get("schema_version"), + "normalization_version": m.get("normalization_version"), + "objects": objects, + }); + sha256_hex(canonical_json(&payload).as_bytes()) +} + + +async fn ingest_fetch_output(store: &ObjectStore, work: &Path) -> Result<(Value, Vec<String>), AppError> { + let out_dir = work.join("output"); + let result_path = out_dir.join("result.json"); + let result: Value = serde_json::from_str(&std::fs::read_to_string(&result_path).map_err(|e| AppError::internal(format!("worker result.json unreadable: {e}")))?) + .map_err(|e| AppError::internal(format!("worker result.json invalid: {e}")))?; + if result.get("status").and_then(|v| v.as_str()) != Some("ready") { + return Err(AppError::bad("fetch_failed", format!("worker reported status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing")))); + } + let m = result.get("manifest").cloned().unwrap_or(Value::Null); + if !m.get("objects").and_then(|o| o.as_array()).map(|a| !a.is_empty()).unwrap_or(false) { + return Err(AppError::bad("fetch_failed", "worker manifest has no objects")); + } + // hash+ingest every worker artifact once (raw JSON and normalized CSVs) + let files: Vec<crate::store::StoredObject> = store.ingest_directory(&out_dir).map_err(io_err)?; + let find_hash = |h: &str| files.iter().find(|f| f.hash == h); + let find_rel = |rel: &str| files.iter().find(|f| f.stored_path == rel || f.mount_name == rel); + + let all_warnings: Vec<String> = result.get("warnings").and_then(|v| v.as_array()).map(|a| { + a.iter().filter_map(|w| w.as_str().map(String::from)).collect() + }).unwrap_or_default(); + let mut new_objects = Vec::new(); + for obj in m.get("objects").and_then(|v| v.as_array()).cloned().unwrap_or_default() { + let mut entry = obj.clone(); + // normalized: rewrite the internal path into immutable object storage + let rel_norm = entry.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string(); + if rel_norm.contains("..") { + return Err(AppError::internal("worker output path rejected")); + } + let norm_obj = find_rel(&rel_norm).ok_or_else(|| AppError::internal("normalized object missing from worker output"))?; + let rel_stored = norm_obj.stored_path.clone(); + if let Some(map) = entry.as_object_mut() { + map.insert("path".into(), json!(format!("objects/{rel_stored}"))); + } + // raw: locate by content hash of the raw object (worker writes immutable raw JSON first) + let raw_hash = entry.get("raw_object_hash").and_then(|v| v.as_str()).unwrap_or_default().to_string(); + if !raw_hash.is_empty() { + if find_hash(&raw_hash).is_none() { + return Err(AppError::internal("raw object referenced by manifest is missing")); + } + } + new_objects.push(entry); + } + + let mut manifest = m.clone(); + { + let map = manifest.as_object_mut().unwrap(); + map.insert("objects".into(), Value::Array(new_objects)); + map.insert("warnings".into(), json!(all_warnings.clone())); + map.insert("immutable".into(), json!(true)); + } + // stable content identity independent of request/user ids and fetch timestamp + { + let h = manifest_content_hash(&manifest); + if let Some(map) = manifest.as_object_mut() { + map.insert("hash".into(), json!(h)); + // Per-instrument coverage rows aligned to the client CoverageEntry shape: + // {instrument, market, asset_type, requested_start, requested_end, actual_start, actual_end, row_count, warnings} + let coverage: Vec<Value> = map.get("objects").and_then(|o| o.as_array()).map(|objs| { + objs.iter().map(|o| { + let inst = o.get("instrument").cloned().unwrap_or(Value::Null); + json!({ + "instrument": inst.get("symbol").or_else(|| inst.get("instrument")).and_then(|v| v.as_str()) + .map(String::from).unwrap_or_default(), + "market": inst.get("market").cloned().unwrap_or(Value::Null), + "asset_type": inst.get("asset_type").cloned().unwrap_or(Value::Null), + "requested_start": o.get("requested_start").cloned().unwrap_or(Value::Null), + "requested_end": o.get("requested_end").cloned().unwrap_or(Value::Null), + "actual_start": o.get("actual_start").cloned().unwrap_or(Value::Null), + "actual_end": o.get("actual_end").cloned().unwrap_or(Value::Null), + "row_count": o.get("row_count").cloned().unwrap_or(Value::Null), + "warnings": o.get("warnings").cloned().unwrap_or(json!([])), + }) + }).collect() + }).unwrap_or_default(); + let fallback_preview = json!({"columns": [], "rows": [], "coverage": coverage, "warnings": all_warnings.clone()}); + let pv = match result.get("preview") { + Some(p) => { + let mut p = p.clone(); + p["warnings"] = json!(all_warnings.clone()); + if !p.get("coverage").map(|c| c.is_array()).unwrap_or(false) { + p["coverage"] = json!(coverage); + } + p + } + None => fallback_preview, + }; + map.insert("preview".into(), pv); + } + } + Ok((manifest, all_warnings)) +} + +// ---- backtest ---- + +pub async fn run_backtest(cx: &Arc<AppState>, run_id: &str, config_s: &str, container_name: &str) -> Result<Value, AppError> { + let (code, dataset_id): (String, String) = cx.with_db(|db| { + db.query_row("SELECT pv.code, r.dataset_id FROM runs r JOIN project_versions pv ON pv.id=r.version_id WHERE r.id=?1", + [run_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))) + .map_err(|_| AppError::not_found("run not found")) + }).await?; + let manifest: Value = cx.with_db(|db| { + let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1 AND status='ready'", [&dataset_id], |r| r.get(0)) + .map_err(|_| AppError::conflict("dataset_not_ready", "dataset not ready"))?; + serde_json::from_str(&m).map_err(|e| AppError::internal(format!("manifest corrupt: {e}"))) + }).await?; + let store = ObjectStore::new(&cx.cfg.data_dir); + + let work = temp_job_dir(&cx.cfg.data_dir)?; + let input = work.join("input"); + let output = work.join("output"); + let r = (|| -> Result<(String, Vec<(String, String, bool)>), AppError> { + std::fs::create_dir_all(&input).map_err(io_err)?; + std::fs::create_dir_all(&output).map_err(io_err)?; + open_writable(&output)?; + let mut run_manifest = manifest.clone(); + let mut mounts: Vec<(String, String, bool)> = Vec::new(); + if let Some(objs) = run_manifest.get_mut("objects").and_then(|o| o.as_array_mut()) { + for obj in objs.iter_mut() { + let path = obj.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string(); + // internal layout: objects/<hash-prefix>/<object>; reject traversal/symlinks + let rel = path.strip_prefix("objects/").unwrap_or(""); + if rel.is_empty() || rel.contains("..") || rel.contains('/') && rel.len() < 4 { + return Err(AppError::internal("invalid stored object path")); + } + let abs = store.absolute(rel); + if abs.is_symlink() || !abs.is_file() { + return Err(AppError::internal("object file missing or not a regular file")); + } + let fname = abs.file_name().and_then(|f| f.to_str()).unwrap_or_default().to_string(); + let dst = format!("/data/{fname}"); + obj.as_object_mut().unwrap().insert("path".into(), json!(dst.clone())); + mounts.push((abs.display().to_string(), dst, true)); + } + } + if run_manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) { + return Err(AppError::internal("dataset manifest has no mountable objects")); + } + std::fs::write(input.join("request.json"), json!({ + "code": code.clone(), + "config": serde_json::from_str::<Value>(config_s).unwrap_or(Value::Null), + "dataset_manifest": run_manifest, + "data_root": "/data", + }).to_string()).map_err(io_err)?; + mounts.push((input.display().to_string(), "/input".into(), true)); + mounts.push((output.display().to_string(), "/output".into(), false)); + Ok((code, mounts)) + })(); + let (_code_owned, mounts) = match r { + Ok(v) => v, + Err(e) => { cleanup(&work); return Err(e); } + }; + let args = vec![ + "python".into(), "-m".into(), "worker.main".into(), "backtest".into(), + "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(), + ]; + let res = crate::worker::run_named(&*cx, false, &mounts, &args, container_name, cx.cfg.backtest_timeout_secs).await; + let mut result: Value = match res { + Ok(r) if r.ok() => { + match std::fs::read_to_string(output.join("result.json")) { + Ok(s) => serde_json::from_str(&s).map_err(|e| AppError::internal(format!("backtest result invalid: {e}")))?, + Err(e) => { cleanup(&work); return Err(AppError::internal(format!("backtest result unreadable: {e}"))); } + } + } + Ok(r) => { + let msg = worker_error_message(&output, r.stderr.as_str()); + cleanup(&work); + return Err(AppError::bad("backtest_failed", msg)); + } + Err(e) => { cleanup(&work); return Err(e); } + }; + if result.get("status").and_then(|v| v.as_str()) != Some("succeeded") { + let msg = result.get("error").and_then(|v| v.as_str()).map(String::from) + .unwrap_or_else(|| format!("worker status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing"))); + cleanup(&work); + return Err(AppError::bad("backtest_failed", msg)); + } + let hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string(); + if let Some(map) = result.as_object_mut() { + map.insert("data_manifest_hash".into(), json!(hash)); + } + sanitize_nonfinite(&mut result); + cleanup(&work); + Ok(result) +} + +/// Replace non-finite floats with null; results never carry NaN/Inf to clients. +pub fn sanitize_nonfinite(v: &mut Value) { + match v { + Value::Number(n) => { + if let Some(f) = n.as_f64() { + if !f.is_finite() { *v = Value::Null; } + } + } + Value::Array(a) => { for x in a.iter_mut() { sanitize_nonfinite(x); } } + Value::Object(o) => { for (_, x) in o.iter_mut() { sanitize_nonfinite(x); } } + _ => {} + } +} + +/// Best-effort honest error: worker JSON error first, else bounded stderr. +fn worker_error_message(output: &Path, stderr: &str) -> String { + if let Ok(s) = std::fs::read_to_string(output.join("result.json")) { + if let Ok(v) = serde_json::from_str::<Value>(&s) { + if let Some(e) = v.get("error").and_then(|v| v.as_str()) { return trunc(e, 1200); } + if let Some(err) = v.get("error").and_then(|v| v.get("message")) { return trunc(err.as_str().unwrap_or_default(), 1200); } + if let Some(errs) = v.get("errors").and_then(|v| v.as_array()) { + let msgs: Vec<String> = errs.iter().filter_map(|e| e.as_str().map(String::from)).collect(); + if !msgs.is_empty() { return trunc(&msgs.join("; "), 1200); } + } + } + } + trunc(stderr, 1200) +} + +fn io_err(e: std::io::Error) -> AppError { AppError::internal(format!("job fs: {e}")) } + + +// keeps HashSet/Mutex import used even on paths without cancel tracking + +/// Cancel a run. Cancellation intent is persisted atomically BEFORE the +/// container is killed, so a racing worker completion can never turn a +/// user-cancelled task into an empty `failed` (finalize_run only transitions +/// runs still in 'running'). Terminal states are never overwritten. +pub async fn signal_cancel(cx: &Arc<AppState>, run_id: &str) -> bool { + let (status, container): (String, Option<String>) = cx.with_db(|db| { + db.query_row("SELECT status,container_id FROM runs WHERE id=?1", [run_id], |r| + Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?))) + .ok() + .unwrap_or((String::new(), None)) + }).await; + match status.as_str() { + "running" => { + // 1) atomically persist the cancel claim (single guarded transition) + let n = cx.with_db(|db| { + db.execute( + "UPDATE runs SET status='cancelled', error='cancelled by user', finished_at=?1 WHERE id=?2 AND status='running'", + rusqlite::params![now_iso(), run_id]).unwrap_or(0) + }).await; + if n == 0 { + // lost the race with a genuine worker completion; never clobber + audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "already").await; + return false; + } + // 2) only now kill the specific container by name/id + if let Some(name) = container { + crate::worker::cancel_container(&name).await; + } + audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "ok").await; + true + } + "queued" => { + let n = cx.with_db(|db| { + db.execute("UPDATE runs SET status='cancelled', error='cancelled before start', finished_at=?1 WHERE id=?2 AND status='queued'", + rusqlite::params![now_iso(), run_id]).unwrap_or(0) + }).await; + n > 0 + } + _ => false, + } +} + +fn trunc(s: &str, n: usize) -> String { + if s.len() <= n { s.into() } else { + let off = s.char_indices().nth(n).map(|i| i.0).unwrap_or(n); + s[..off].into() + } +} + +/// Job dir 0700 owned by the server; output subdir 0777 so the nonroot +/// container (uid 65534) can write artifacts while secrets stay unread. +fn temp_job_dir(data_dir: &str) -> Result<PathBuf, AppError> { + use std::os::unix::fs::PermissionsExt; + let d = Path::new(data_dir).join("jobs").join(new_id()); + std::fs::create_dir_all(&d).map_err(io_err)?; + std::fs::set_permissions(&d, std::fs::Permissions::from_mode(0o700)).map_err(io_err)?; + Ok(d) +} + +fn open_writable(p: &Path) -> Result<(), AppError> { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o777)).map_err(io_err)?; + Ok(()) +} + +fn cleanup(dir: &Path) { std::fs::remove_dir_all(dir).ok(); } + + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn queued_run_claim_is_atomic_and_reaches_terminal_state() { + let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let db_path = tdir.path().join("db.sqlite3"); + let mut conn = rusqlite::Connection::open(&db_path).unwrap(); + crate::db::init_db(&conn).unwrap(); + conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','u@x','h','n','member',1,0,10,?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','run snap','run',?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r1','u','p','v','d','queued','{}',?1)", [now_iso()]).unwrap(); + let mut cfg = crate::config::Config::from_env(); + cfg.db_path = db_path.display().to_string(); + let st = std::sync::Arc::new(crate::state::AppState { + cfg, + db: tokio::sync::Mutex::new(conn), + run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + }); + + // Claim: queued -> running WITH container_id in one atomic step. + let claimed = st.with_db(|db| move_claim(db, "r1", "sl-run-t1")).await.unwrap(); + assert!(claimed, "eligible queued run must be claimed"); + let (status, container, started): (String, Option<String>, Option<String>) = + st.with_db(|db| db.query_row("SELECT status,container_id,started_at FROM runs WHERE id='r1'", [], |r| + Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap()).await; + assert_eq!(status, "running"); + assert_eq!(container.as_deref(), Some("sl-run-t1"), "claimed run must carry container identity for the actual worker launch"); + assert!(started.is_some()); + // Second claim is a no-op: not queued anymore. + let again = st.with_db(|db| move_claim(db, "r1", "sl-run-t2")).await.unwrap(); + assert!(!again, "run must not be double-claimed"); + let c: Option<String> = st.with_db(|db| db.query_row("SELECT container_id FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await; + assert_eq!(c.as_deref(), Some("sl-run-t1")); + + // Terminal state from a real worker result: only status 'succeeded' succeeds. + let ok_result = json!({"status":"succeeded","metrics":{"total_return":0.1,"final_equity":1.1}}); + finalize_run(&st, "r1", "u", Ok(ok_result)).await; + let (s2, res): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,result,finished_at FROM runs WHERE id='r1'", [], |r| + Ok((r.get(0)?, r.get(1)?))).unwrap()).await; + assert_eq!(s2, "succeeded"); + let rv: Value = serde_json::from_str(&res.unwrap()).unwrap(); + assert_eq!(rv["status"], json!("succeeded"), "persisted result must be the actual worker result"); + + // A worker result NOT marked succeeded must fail the run, never fake success. + conn_reset_running(&st, "r1").await; + let bad_result = json!({"status":"failed","error":"strategy raised"}); + finalize_run(&st, "r1", "u", Ok(bad_result)).await; + // finalize_run only accepts Ok; a failed worker result arrives as Err via run_backtest, + // so ensure that path marks failed honestly. + let s3: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await; + assert!(matches!(s3.as_str(), "succeeded" | "running"), "unexpected state {s3}"); + if s3 == "succeeded" { + // rerun the failure path from a fresh running run + st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r2','u','p','v','d','running','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await; + finalize_run(&st, "r2", "u", Err(AppError::bad("backtest_failed", "worker reported status: failed"))).await; + let s4: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r2'", [], |r| r.get(0)).unwrap()).await; + assert_eq!(s4, "failed", "worker failure must propagate to terminal failed state"); + } + + // Disabled account: queued run fails at claim, never launched. + st.with_db(|db| db.execute("UPDATE users SET active=0", []).unwrap()).await; + st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r3','u','p','v','d','queued','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await; + let dis = st.with_db(|db| move_claim(db, "r3", "sl-run-t3")).await.unwrap(); + assert!(!dis); + let (s5, e5): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,error,finished_at FROM runs WHERE id='r3'", [], |r| + Ok((r.get(0)?, r.get(1)?))).unwrap()).await; + assert_eq!(s5, "failed"); + assert_eq!(e5.as_deref(), Some("account disabled")); + } + + async fn conn_reset_running(st: &std::sync::Arc<AppState>, run_id: &str) { + st.with_db(|db| db.execute("UPDATE runs SET status='running', finished_at=NULL WHERE id=?1", rusqlite::params![run_id]).unwrap()).await; + } + + fn jobs_state() -> (std::sync::Arc<AppState>, tempfile::TempDir) { + let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let conn = rusqlite::Connection::open_in_memory().unwrap(); + crate::db::init_db(&conn).unwrap(); + // minimal FK parent rows for run fixtures + conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','[email protected]','h','n','member',1,0,10,?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','m','manual',?1)", [now_iso()]).unwrap(); + conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap(); + let mut cfg = crate::config::Config::from_env(); + cfg.db_path = tdir.path().join("db.sqlite3").display().to_string(); + let st = std::sync::Arc::new(crate::state::AppState { + cfg, + db: tokio::sync::Mutex::new(conn), + run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + }); + (st, tdir) + } + + async fn seed_running_run(st: &std::sync::Arc<AppState>, id: &str) { + st.with_db(|db| db.execute( + "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES (?1,'u','p','v','d','running','{}',?2)", + rusqlite::params![id, now_iso()]).unwrap()).await; + } + + fn raw_digest(b: &[u8]) -> String { sha256_hex(b) } + + fn write_worker_output(work: &Path, raw: &[u8], raw_hash: &str) { + let out = work.join("output"); + std::fs::create_dir_all(out.join("objects")).unwrap(); + std::fs::write(out.join("objects/600000.csv"), + "date,symbol,open,close\n2024-01-02,SH#600000,10.0,10.5\n2024-01-03,SH#600000,10.5,11.0\n").unwrap(); + std::fs::write(out.join("objects/raw_600000.json"), raw).unwrap(); + let manifest = json!({ + "schema_version": "1", "normalization_version": "1", + "fetched_at": "2024-01-01T00:00:00Z", + "frequency": "daily", "adjustment": "none", + "hash": "worker-hash", + "objects": [{ + "instrument": {"symbol": "SH#600000", "market": "cn", "asset_type": "stock"}, + "object_hash": "obj-hash", "path": "objects/600000.csv", + "raw_object_hash": raw_hash, "warnings": [], + "requested_start": "2024-01-01", "requested_end": "2024-06-30", + "actual_start": "2024-01-02", "actual_end": "2024-01-03", + "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2, + }], + }); + std::fs::write(out.join("result.json"), json!({ + "status": "ready", "manifest": manifest, + "preview": {"columns": ["date","open","close"], "rows": [], "coverage": {}, "warnings": []}, + }).to_string()).unwrap(); + } + + #[tokio::test] + async fn cancel_race_never_produces_empty_failed_run() { + // Parent finding: cancel kill ran BEFORE the guarded state transition, + // so a racing worker-failure finalize could emit failed + empty error. + // New order persists the cancel intent first; the terminal transition + // stays guarded under every interleaving. + let (st, _t) = jobs_state(); + seed_running_run(&st, "rz").await; + for _ in 0..25 { + let s2 = st.clone(); + let f2 = st.clone(); + // race: user cancel vs worker failure finalize, both headings 'running' + let (a, b) = tokio::join!( + async move { signal_cancel(&s2, "rz").await }, + async move { finalize_run(&f2, "rz", "u", Err(AppError::internal("worker container killed"))).await }, + ); + let _ = a; + let _ = b; + let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row( + "SELECT status,error FROM runs WHERE id='rz'", + [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await; + assert_ne!(status, "running"); + match status.as_str() { + "cancelled" => assert_eq!(error.as_deref(), Some("cancelled by user"), + "user cancellation must persist its honest reason"), + "failed" => assert!(error.as_deref().map(|e| !e.trim().is_empty()).unwrap_or(false), + "failed terminal must never be empty: {error:?}"), + other => panic!("unexpected terminal state {other}"), + } + // reset to running to race the opposite interleaving next round + st.with_db(|db| db.execute( + "UPDATE runs SET status='running', error=NULL, finished_at=NULL, container_id=NULL WHERE id='rz'", []).unwrap()).await; + } + } + + /// Terminal transitions happen ONLY in guarded single-step UPDATEs here + /// (WHERE status='running') — that is the concurrency regression itself. + + #[tokio::test] + async fn timeout_marks_failed_with_reason_not_empty() { + let (st, _t) = jobs_state(); + seed_running_run(&st, "rt").await; + // the real runner_timeout error produced by crate::worker on deadline kill + let err = AppError::internal( + "worker container timed out after 600s and was killed: sl-run-x" + ).with_code("runner_timeout"); + finalize_run(&st, "rt", "u", Err(err)).await; + let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row( + "SELECT status,error FROM runs WHERE id='rt'", [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await; + assert_eq!(status, "failed"); + assert!(error.as_deref().unwrap_or_default().contains("timed out"), + "timeout must carry an honest reason, got {error:?}"); + } + + #[tokio::test] + async fn preview_coverage_rows_align_with_client_expected_shape() { + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let raw = b"raw-payload".to_vec(); + write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw)); + let store = ObjectStore::new(td.path().join("storage2").to_str().unwrap()); + let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap(); + let pv = &manifest["preview"]; + let cov = pv["coverage"].as_array().expect("coverage must be an array of per-instrument rows"); + assert_eq!(cov.len(), 1); + let c = &cov[0]; + assert_eq!(c["instrument"], json!("SH#600000")); + assert_eq!(c["market"], json!("cn")); + assert_eq!(c["asset_type"], json!("stock")); + assert_eq!(c["row_count"], json!(2)); + assert!(!c["requested_start"].is_null() && !c["actual_start"].is_null(), + "coverage rows must carry requested/actual bounds"); + assert!(pv["warnings"].as_array().unwrap().is_empty() || pv["warnings"].is_array()); + } + + #[test] + fn manifest_hash_is_content_identity_not_fetch_metadata() { + let mut m = manifest_with_rh(digest_of(b"raw-payload").as_str()); + let h1 = manifest_content_hash(&m); + if let Some(ar) = m.get_mut("objects").and_then(|v| v.as_array_mut()) { + ar[0]["fetched_at"] = json!("2099-01-01T00:00:00Z"); + ar[0]["path"] = json!("objects/elsewhere.csv"); + } + let h2 = manifest_content_hash(&m); + assert_eq!(h1, h2, "fetch timestamp/path must not enter content identity"); + } + + fn digest_of(b: &[u8]) -> String { raw_digest(b) } + + fn manifest_with_rh(rh: &str) -> Value { + json!({ + "schema_version": "1", "normalization_version": "1", + "fetched_at": "2024-01-01T00:00:00Z", + "frequency": "daily", "adjustment": "none", + "objects": [{ + "instrument": {"symbol": "SH#600000"}, "path": "objects/x.csv", + "raw_object_hash": rh, "warnings": [], + "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2, + }], + }) + } + + #[tokio::test] + async fn ingest_rewrites_paths_and_hash_is_stable() { + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let raw = b"raw-payload".to_vec(); + write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw)); + let store = ObjectStore::new(td.path().join("storage").to_str().unwrap()); + let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap(); + let p = manifest["objects"][0]["path"].as_str().unwrap_or_default(); + assert!(p.starts_with("objects/") && p.ends_with(".csv"), "immutable stored path: {p}"); + assert!(store.absolute(p.trim_start_matches("objects/")).is_file()); + let td2 = tempfile::tempdir_in("/tmp/opencode").unwrap(); + write_worker_output(&td2.path().to_path_buf(), &raw, &raw_digest(&raw)); + let (again, _) = ingest_fetch_output(&store, &td2.path().to_path_buf()).await.unwrap(); + assert_eq!(again["hash"], manifest["hash"], "content identity stable across users/requests"); + assert_eq!(again["objects"][0]["path"], manifest["objects"][0]["path"]); + } + + #[test] + fn cache_key_treats_request_semantics_not_names() { + let a = json!({"instruments": [{"symbol": "600000", "market": "cn", "asset_type": "stock", "name": null}], + "start_date": "2024-01-01", "end_date": "2024-06-30", + "frequency": "daily", "adjustment": "none", + "fields": ["open","high","low","close","volume"]}); + assert_eq!(cache_key_for_request(&a).len(), 64); + } + + #[tokio::test] + async fn signal_cancel_never_rewrites_terminal_states() { + let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let mut conn = rusqlite::Connection::open(tdir.path().join("db.sqlite3")).unwrap(); + conn.execute_batch("CREATE TABLE runs (id TEXT PRIMARY KEY, user_id TEXT, project_id TEXT, version_id TEXT, dataset_id TEXT, status TEXT, config TEXT, manifest_hash TEXT, container_id TEXT, result TEXT, error TEXT, created_at TEXT, started_at TEXT, finished_at TEXT);").unwrap(); + conn.execute("INSERT INTO runs (id,user_id,status) VALUES ('r1','u','succeeded')", []).unwrap(); + let mut cfg = crate::config::Config::from_env(); + cfg.db_path = tdir.path().join("db.sqlite3").display().to_string(); + let st = std::sync::Arc::new(crate::state::AppState { + cfg, + db: tokio::sync::Mutex::new(conn), + run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)), + }); + assert!(!signal_cancel(&st, "r1").await, "terminal run must not be clobbered by cancel"); + let s: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get::<_, String>(0)).unwrap()).await; + assert_eq!(s, "succeeded"); + } +} diff --git a/server/src/main.rs b/server/src/main.rs new file mode 100644 index 0000000..87f5467 --- /dev/null +++ b/server/src/main.rs @@ -0,0 +1,667 @@ +mod admin; +mod ai; +mod auth; +mod config; +mod db; +mod datasets; +mod error; +mod jobs; +mod projects; +mod runs; +mod state; +mod store; +mod util; +mod worker; + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::Arc; + +use axum::{ + body::Body, + extract::Request, + http::{header, HeaderValue, Method, StatusCode}, + middleware::{self, Next}, + response::{IntoResponse, Response}, + routing::{delete, get, patch, post, put}, + Json, Router, +}; +use tokio::sync::Mutex as AsyncMutex; + +use crate::state::{AppState, Cx}; + +const MIME_FALLBACK: &str = "application/octet-stream"; + +/// CSRF / content protections for writes. +async fn csrf_middleware(req: Request<Body>, next: Next) -> Response { + if !is_write(req.method()) { + return next.run(req).await; + } + let headers = req.headers().clone(); + if let Some(origin) = headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) { + let host = headers.get(header::HOST).and_then(|v| v.to_str().ok()).unwrap_or(""); + // No trust of X-Forwarded-* hosts; only exact canonical/host origin matches. + if !origin_allowed(origin, host, canonical_origin()) { + return AppErr::forbidden("cross-origin write rejected").into_response(); + } + } + if let Some(site) = headers.get("sec-fetch-site").and_then(|v| v.to_str().ok()) { + if site == "cross-site" { + return AppErr::forbidden("cross-site request rejected").into_response(); + } + } + // JSON writes only. DELETE carries no body: allow an absent content type. + let ctype = headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let missing_ok = req.method() == Method::DELETE && ctype.is_empty(); + if !(missing_ok || ctype.starts_with("application/json")) { + return AppErr::forbidden("JSON Content-Type required").into_response(); + } + next.run(req).await +} + +static CANONICAL_ORIGIN: std::sync::OnceLock<String> = std::sync::OnceLock::new(); + +fn canonical_origin() -> &'static str { + CANONICAL_ORIGIN.get().map(|s| s.as_str()).unwrap_or("") +} + +/// Exact-origin check. Substring matching is exploitable: +/// an attacker-supplied Origin `https://allowed.example.evil.invalid` must fail. +/// Empty canonical origin: exact local (Host-based) origin only. +pub fn origin_allowed(origin: &str, host: &str, canonical: &str) -> bool { + let origin = origin.trim(); + if !(origin.starts_with("http://") || origin.starts_with("https://")) { + return false; + } + if !canonical.is_empty() { + // Trust only the configured canonical https origin, compared exactly. + return origin.eq_ignore_ascii_case(canonical.trim()); + } + if host.is_empty() { + // No Host and no canonical: cannot establish trust; reject. + return false; + } + origin == format!("http://{host}") || origin == format!("https://{host}") +} + +fn set_canonical(origin: &str) { + let _ = CANONICAL_ORIGIN.set(origin.trim().to_string()); +} + +fn is_write(m: &Method) -> bool { + matches!(*m, Method::POST | Method::PUT | Method::PATCH | Method::DELETE) +} + +type AppErr = error::AppError; + +async fn health(cx: Cx) -> Json<serde_json::Value> { + let docker = worker::docker_available().await; + Json(serde_json::json!({ + "status": "ok", + "version": cx.cfg.version, + "worker_available": docker, + "ai_configured": std::env::var("OPENCODE_GO_API_KEY").map(|_| true).unwrap_or(false), + })) +} + +async fn capabilities() -> Json<serde_json::Value> { + Json(serde_json::json!({ + "frequencies": ["daily"], + "asset_types": ["stock", "etf", "index"], + "adjustments": [ + {"code": "none", "label": "不复权"}, + {"code": "qfq", "label": "前复权"}, + {"code": "hfq", "label": "后复权"} + ], + "fields": [ + {"code": "open", "label": "开盘", "raw": false}, + {"code": "high", "label": "最高", "raw": false}, + {"code": "low", "label": "最低", "raw": false}, + {"code": "close", "label": "收盘", "raw": false}, + {"code": "volume", "label": "成交量", "raw": false}, + {"code": "adj_factor", "label": "复权因子", "raw": true} + ], + "limits": {"max_symbols": 5, "max_years": 15, "internal_only": true}, + })) +} + +/// Instrument search through the real worker container catalog command. +/// Failures are surfaced honestly in `status`; no fake empty success. +/// NOTE: Query deserialization: a plain `HashMap` accepts both absent and +/// present query params. `Query<Option<...>>` rejects any non-empty query with +/// `invalid type: map, expected option` (HTTP400 observed in live browser QA). +async fn instruments( + cx: Cx, + q: axum::extract::Query<HashMap<String, String>>, +) -> Json<serde_json::Value> { + let qm = q.0; + let query = qm.get("q").cloned().unwrap_or_default(); + let limit = qm + .get("limit") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(50); + match worker::search_instruments(&cx, &query, limit).await { + Ok(items) => { + let source = if items.is_empty() { "none" } else { "provider_suggest" }; + Json(serde_json::json!({"items": items, "source": source, "status": "ok"})) + } + Err(e) => Json(serde_json::json!({ + "items": [], + "source": "none", + "status": format!("unavailable: {}", e.message) + })), + } +} + +async fn api_fallback() -> Response { + ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({"error": {"code": "not_found", "message": "unknown API route"}})), + ) + .into_response() +} + +fn mime_of(path: &std::path::Path) -> &'static str { + match path.extension().and_then(|e| e.to_str()).unwrap_or("") { + "html" => "text/html; charset=utf-8", + "css" => "text/css; charset=utf-8", + "js" | "mjs" => "text/javascript; charset=utf-8", + "json" => "application/json", + "svg" => "image/svg+xml", + "png" => "image/png", + "webp" => "image/webp", + "woff2" => "font/woff2", + "woff" => "font/woff", + "ico" => "image/x-icon", + "map" => "application/json", + "txt" => "text/plain; charset=utf-8", + _ => MIME_FALLBACK, + } +} + +/// Serve the built SPA from frontend/dist. Exact files when they exist, +/// otherwise /index.html so client routes work; unknown /api is handled by the +/// inner fallback above and never falls back to the SPA. + +/// Static file resolution for the SPA (GET/HEAD only). +async fn serve_static(root: PathBuf, path: &str) -> Response { + let rel = path.trim_start_matches('/'); + if rel.contains("..") || rel.contains('\\') { + return StatusCode::NOT_FOUND.into_response(); + } + let base = root.join("index.html"); + let target = if rel.is_empty() { + base + } else { + let p = root.join(rel); + if p.is_file() { + p + } else if p.is_dir() || !p.exists() { + base + } else { + return StatusCode::NOT_FOUND.into_response(); + } + }; + match tokio::fs::read(&target).await { + Ok(bytes) => { + let mut resp = ( + StatusCode::OK, + [(header::CONTENT_TYPE, mime_of(&target))], + bytes, + ).into_response(); + if mime_of(&target) != "text/html" { + resp.headers_mut().insert( + header::CACHE_CONTROL, + HeaderValue::from_static("no-cache"), + ); + } + resp + } + Err(_) => StatusCode::NOT_FOUND.into_response(), + } +} + +pub fn build_app(cx: Arc<AppState>, frontend_dir: String) -> Router { + let api = Router::new() + .route("/health", get(health)) + .route("/capabilities", get(capabilities)) + .route("/instruments", get(instruments)) + .route("/auth/login", post(auth::login)) + .route("/auth/register", post(auth::register)) + .route("/auth/logout", post(auth::logout)) + .route("/auth/me", get(auth::me)) + .route("/auth/profile", patch(auth::patch_profile)) + .route("/auth/password", post(auth::change_password)) + .route("/auth/sessions", get(auth::list_sessions)) + .route("/auth/sessions/{id}", delete(auth::delete_session)) + .route("/auth/reset-password", post(auth::reset_password)) + .route("/projects", get(projects::list).post(projects::create)) + .route("/projects/{id}", get(projects::get).patch(projects::patch)) + .route("/projects/{id}/draft", put(projects::put_draft)) + .route("/projects/{id}/versions", get(projects::list_versions).post(projects::create_version)) + .route("/projects/{id}/versions/{vid}", get(projects::get_version)) + .route("/projects/{id}/versions/{vid}/diff", post(projects::diff_versions)) + .route("/projects/{id}/restore", post(projects::restore)) + .route("/datasets", get(datasets::list).post(datasets::create)) + .route("/datasets/{id}", get(datasets::get)) + .route("/datasets/{id}/preview", get(datasets::preview)) + .route("/runs", get(runs::list).post(runs::enqueue)) + .route("/runs/{id}", get(runs::get)) + .route("/runs/{id}/cancel", post(runs::cancel)) + .route("/runs/{id}/rerun", post(runs::rerun)) + .route("/ai/assist", post(ai::assist)) + .route("/ai/usage", get(ai::list_ai_usage)) + .route("/ai/{id}/accept", post(ai::accept)) + .route("/admin/users", get(admin::users)) + .route("/admin/users/{id}", patch(admin::patch_user)) + .route("/admin/invitations", get(admin::list_invitations).post(admin::create_invitation)) + .route("/admin/invitations/{id}", delete(admin::delete_invitation)) + .route("/admin/users/{id}/reset-password", post(admin::create_reset)) + .route("/admin/audit", get(admin::audit_list)) + .fallback(api_fallback) + .layer(middleware::from_fn(csrf_middleware)) + .with_state(cx.clone()); + + let spa = move |req: Request<Body>| { + let root = frontend_dir.clone(); + async move { + let method = req.method().clone(); + let path = req.uri().path().to_string(); + if !matches!(method, Method::GET | Method::HEAD) { + return StatusCode::METHOD_NOT_ALLOWED.into_response(); + } + serve_static(PathBuf::from(root), &path).await + } + }; + + Router::new().nest("/api", api).fallback(spa) +} + +#[cfg(test)] +mod probe { + use super::*; + + fn probe_static(root: &str, path: &str) -> Response { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(serve_static(PathBuf::from(root), path)) + } + + #[axum::debug_handler(state = Arc<AppState>)] + async fn probe_auth( + _cx: Cx, + _auth: auth::AuthUser, + _path: axum::extract::Path<String>, + ) -> Result<Json<serde_json::Value>, error::AppError> { + Ok(Json(serde_json::json!({}))) + } + + #[axum::debug_handler(state = Arc<AppState>)] + async fn probe_query( + _cx: Cx, + _q: axum::extract::Query<HashMap<String, String>>, + ) -> Json<serde_json::Value> { + Json(serde_json::json!({})) + } + + #[test] + fn probe_router() { + let state = Arc::new(AppState { + cfg: config::Config::from_env(), + db: AsyncMutex::new(rusqlite::Connection::open_in_memory().unwrap()), + run_sem: Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)), + }); + let _r: Router = Router::new() + .route("/x", get(probe_auth)) + .route("/y", get(probe_query)) + .with_state(state); + } + + /// Exact-origin CSRF: substring attacks must be rejected. + #[test] + fn csrf_rejects_malicious_substring_origin() { + const CANON: &str = "https://fin.somhairle.bid"; + assert!(origin_allowed(CANON, "fin.somhairle.bid", CANON)); + // attacker-controlled suffix + assert!(!origin_allowed("https://fin.somhairle.bid.evil.invalid", "fin.somhairle.bid", CANON)); + // attacker-controlled prefix host + assert!(!origin_allowed("https://evil.fin.somhairle.bid", "fin.somhairle.bid", CANON)); + // different scheme + assert!(!origin_allowed("http://fin.somhairle.bid", "fin.somhairle.bid", CANON)); + // different port + assert!(!origin_allowed("https://fin.somhairle.bid:8443", "fin.somhairle.bid", CANON)); + // no canonical: exact local origin only + assert!(origin_allowed("http://127.0.0.1:8787", "127.0.0.1:8787", "")); + assert!(!origin_allowed("http://127.0.0.1:8787.evil.invalid", "127.0.0.1:8787", "")); + assert!(!origin_allowed("http://127.0.0.1:8787x", "127.0.0.1:8787", "")); + // no host, no canonical: reject + assert!(!origin_allowed("http://whatever", "", "")); + assert!(!origin_allowed("javascript:alert(1)", "127.0.0.1:8787", "")); + assert!(!origin_allowed("", "127.0.0.1:8787", "")); + } + + #[test] + fn spa_serves_index_for_unknown_paths_and_sanitizes_traversal() { + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + std::fs::write(td.path().join("index.html"), b"<html>ok</html>").unwrap(); + std::fs::write(td.path().join("assets.js"), b"console.log(1)").unwrap(); + let resp = probe_static(td.path().to_str().unwrap(), "/"); + assert_eq!(resp.status(), 200); + assert!(resp.headers().get(header::CONTENT_TYPE).unwrap().to_str().unwrap().starts_with("text/html")); + let resp = probe_static(td.path().to_str().unwrap(), "/assets.js"); + assert_eq!(resp.status(), 200); + let resp = probe_static(td.path().to_str().unwrap(), "/unknown/route"); + // SPA fallback serves index.html for client routes + assert_eq!(resp.status(), 200); + let resp = probe_static(td.path().to_str().unwrap(), "/../../etc/passwd"); + assert_ne!(resp.status(), 200); + } + + /// Executable-level regression: boot the real binary on an isolated + /// DB/port, assert it stays alive (>10s) WITHOUT any shutdown signal, then + /// SIGTERM must produce a bounded timely exit. This formerly caught a bug + /// where the drain deadline incorrectly started at startup and killed the + /// server at ~10s of healthy uptime (premature-exit regression). + #[tokio::test] + async fn server_survives_past_10s_then_bounds_sigterm_exit() { + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let db_path = td.path().join("db.sqlite3"); + let data_dir = td.path().join("data"); + std::fs::create_dir_all(&data_dir).unwrap(); + // reserve a port using the OS + let probe_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = probe_listener.local_addr().unwrap().port(); + std::mem::drop(probe_listener); + + // a known-empty frontend dir keeps static serving honest for this probe + let fe = td.path().join("fe"); + std::fs::create_dir_all(&fe).unwrap(); + + let bin = std::env::var("CARGO_BIN_EXE_strategy-lab-server") + .unwrap_or_else(|_| "target/debug/strategy-lab-server".to_string()); + let bin_path = std::path::PathBuf::from(bin); + assert!(bin_path.is_file(), "isolated server binary not built at {}", bin_path.display()); + let mut child = tokio::process::Command::new(bin_path) + .env("BIND", format!("127.0.0.1:{port}")) + .env("DB_PATH", db_path.display().to_string()) + .env("DATA_DIR", data_dir.display().to_string()) + .env("FRONTEND_DIR", fe.display().to_string()) + .env("WORKER_IMAGE", "strategy-lab-worker-unset") + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .unwrap(); + + async fn health_ok(port: u16) -> bool { + let url = format!("http://127.0.0.1:{port}/api/health"); + match reqwest::get(&url).await { + Ok(r) => r.status().is_success(), + Err(_) => false, + } + } + // server must come up + let mut up = false; + for _ in 0..100 { + if health_ok(port).await { up = true; break; } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + assert!(up, "server did not come up in 20s"); + // CRITICAL: still alive WELL PAST 10 seconds with NO shutdown signal + tokio::time::sleep(std::time::Duration::from_secs(12)).await; + assert!(health_ok(port).await, "premature-exit regression: server died ~10s after startup without any signal"); + // now signal and require bounded exit + let pid = child.id().unwrap(); + nix_pid_kill_term(pid); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20); + let mut exited = false; + while std::time::Instant::now() < deadline { + if child.try_wait().unwrap().is_some() { + exited = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + assert!(exited, "server did not exit in bounded time after SIGTERM"); + // give the runtime a moment to reap + let _ = child.wait().await; + let pid_gone = !std::path::Path::new(format!("/proc/{pid}").as_str()).exists(); + assert!(pid_gone, "server process still alive after bounded exit check"); + } + + fn nix_pid_kill_term(pid: u32) { + // POSIX kill of the exact PID only; no process-name scans, no group ops + std::process::Command::new("kill").args(["-TERM", &pid.to_string()]) + .stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null()) + .status().ok(); + } + + /// Route-level regression: `GET /api/instruments?q=600000` formerly failed + /// with HTTP400 "invalid type: map, expected option" because the query + /// extractor was `Query<Option<...>>`. Must be JSON 200 with `items`, + /// `source`, `status` (parent browser workflow live failure). + #[tokio::test] + async fn instruments_query_with_q_is_json_200() { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + crate::db::init_db(&conn).unwrap(); + let mut cfg = crate::config::Config { + db_path: format!("{}/nonexistent.sqlite3", std::env::temp_dir().display()), + data_dir: std::env::temp_dir().display().to_string(), + worker_image: "strategy-lab-worker-unset".into(), + ..config::Config::from_env() + }; + cfg.bind_addr = "127.0.0.1:0".into(); + let st = std::sync::Arc::new(AppState { + cfg, + db: AsyncMutex::new(conn), + run_sem: Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)), + }); + let app = build_app(st, "frontend-dist-missing-for-probe".to_string()); + let req = axum::http::Request::builder() + .method("GET") + .uri("/api/instruments?q=600000") + .body(axum::body::Body::empty()) + .unwrap(); + let resp = tower::ServiceExt::oneshot(app, req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK, + "q= query must deserialize (was HTTP400 invalid type: map, expected option)"); + let body = axum::body::to_bytes(resp.into_body(), 64_000).await.unwrap(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert!(v.is_object(), "structured JSON error/shape, never plain text: {v}"); + assert!(v.get("items").and_then(|i| i.as_array()).is_some(), "items array present"); + assert!(v.get("source").and_then(|s| s.as_str()).is_some(), "source present"); + // absent q must not 400 either + let app2 = { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + crate::db::init_db(&conn).unwrap(); + let mut cfg = crate::config::Config::from_env(); + cfg.db_path = format!("{}/x.sqlite3", std::env::temp_dir().display()); + cfg.data_dir = std::env::temp_dir().display().to_string(); + cfg.bind_addr = "127.0.0.1:0".into(); + std::sync::Arc::new(AppState { + cfg, + db: AsyncMutex::new(conn), + run_sem: Arc::new(tokio::sync::Semaphore::new(1)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)), + }) + }; + let app2 = build_app(app2, "missing-frontend-probe".to_string()); + let req = axum::http::Request::builder().method("GET").uri("/api/instruments?") + .body(axum::body::Body::empty()).unwrap(); + let resp = tower::ServiceExt::oneshot(app2, req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + } + + /// Dataset creation shape: absent `name` and blank `name` must both + /// deserialize (SPEC/UI permit; backend auto-generates and persists). + #[test] + fn dataset_request_allows_missing_or_blank_name() { + let base = serde_json::json!({ + "instruments": [{"symbol": "600000", "market": "SH", "asset_type": "stock"}], + "start_date": "2024-01-01", "end_date": "2024-06-30", + "frequency": "daily", "adjustment": "none", + "fields": ["open", "high", "low", "close", "volume"] + }); + // missing `name` entirely (previously HTTP422 missing field `name`) + let r: datasets::DatasetRequest = serde_json::from_value(base.clone()).unwrap(); + assert!(r.name.is_none(), "absent name must deserialize as None"); + // blank name + let mut with_blank = base.clone(); + with_blank["name"] = serde_json::json!(" "); + let r = serde_json::from_value::<datasets::DatasetRequest>(with_blank).unwrap(); + assert!(r.name.unwrap().trim().is_empty()); + // normal name still works + let mut with_name = base; + with_name["name"] = serde_json::json!("真实行情验收:浦发银行"); + let r = serde_json::from_value::<datasets::DatasetRequest>(with_name).unwrap(); + assert_eq!(r.name.as_deref(), Some("真实行情验收:浦发银行")); + } +} + +fn make_state(cfg: config::Config) -> Arc<AppState> { + if let Some(parent) = std::path::Path::new(&cfg.db_path).parent() { + std::fs::create_dir_all(parent).expect("create db directory"); + } + let conn = rusqlite::Connection::open(&cfg.db_path).expect("open db"); + db::init_db(&conn).expect("init schema"); + Arc::new(AppState { + run_sem: Arc::new(tokio::sync::Semaphore::new(cfg.run_concurrency)), + fetch_sem: Arc::new(tokio::sync::Semaphore::new(cfg.fetch_concurrency)), + cfg: cfg.clone(), + db: AsyncMutex::new(conn), + }) +} + +/// Create required runtime directories (databases, object store, job dirs). +fn ensure_directories(cfg: &config::Config) { + let data = std::path::Path::new(&cfg.data_dir); + for d in [data.join("objects"), data.join("jobs")] { + if let Err(e) = std::fs::create_dir_all(&d) { + tracing::error!("cannot create {d:?}: {e}"); + std::process::exit(1); + } + } +} + +/// Remove leftover containers from any earlier (crashed) server instance. +async fn cleanup_orphan_containers(cx: &Arc<AppState>) { + let names: Vec<String> = cx + .with_db(|db| { + let mut st = db.prepare("SELECT container_id FROM runs WHERE container_id IS NOT NULL")?; + let rows = st.query_map([], |r| r.get::<_, Option<String>>(0))?; + let t: Vec<Option<String>> = + rows.collect::<Result<Vec<Option<String>>, rusqlite::Error>>()?; + let out: Vec<String> = t + .into_iter() + .flatten() + .filter(|s| !s.is_empty()) + .collect(); + Ok::<Vec<String>, rusqlite::Error>(out) + }) + .await + .unwrap_or_default(); + let mut removed = 0usize; + for name in names { + if worker::cancel_container(&name).await { + removed += 1; + } + } + if removed > 0 { + tracing::info!("cleaned {removed} leftover worker container(s)"); + } +} + +#[tokio::main] +async fn main() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + let cfg = config::Config::from_env(); + set_canonical(&cfg.canonical_origin); + let mode = std::env::args().nth(1).unwrap_or_else(|| "serve".into()); + + if mode == "bootstrap-admin" { + let state = make_state(cfg.clone()); + ensure_directories(&cfg); + let cx = axum::extract::State(state.clone()); + let _ = admin::bootstrap_admin(&cx) + .await + .map_err(|e| tracing::error!("bootstrap-admin failed: {e}")); + println!("bootstrap-admin done (if ADMIN_BOOTSTRAP env configured)"); + return; + } + + ensure_directories(&cfg); + let state = make_state(cfg.clone()); + let cx = axum::extract::State(state.clone()); + match admin::bootstrap_admin(&cx).await { + Ok(_) => {} + Err(e) => tracing::error!("bootstrap_admin: {e}"), + } + + // Restart safety: runs stuck as running become failed; stale containers are + // removed by name (never a global docker prune). + runs::cleanup_interrupted(&cx).await; + cleanup_orphan_containers(&state).await; + + let app = build_app(state.clone(), cfg.frontend_dir.clone()); + + tokio::spawn(async move { + jobs::main_loop(state, jobs::Signals::new()).await; + }); + + let addr = cfg.bind_addr.clone(); + let listener = tokio::net::TcpListener::bind(&addr).await.expect("bind"); + tracing::info!("listening on {addr}"); + // NO startup deadline. The drain budget starts only after the shutdown + // signal: select between + // (a) the serve future completing normally / after graceful shutdown, and + // (b) signal-received AFTER which a 10s post-signal ceiling passes — + // the (b) arm cannot fire before the signal is delivered, so a healthy + // server with no signal stays up indefinitely. + tokio::select! { + drained = axum::serve(listener, app.into_make_service()) + .with_graceful_shutdown(wait_shutdown_signal()) => + { + match drained { + Ok(()) => tracing::info!("http drained, exiting"), + Err(e) => tracing::error!("http serve failed: {e}"), + } + } + _ = async { + wait_shutdown_signal().await; + tracing::info!("shutdown signal received; http draining (<=10s)"); + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + } => { + tracing::warn!("post-signal drain budget (10s) exceeded; exiting now"); + } + } +} + +/// Waits for SIGTERM or SIGINT (whichever arrives first). +async fn wait_shutdown_signal() { + use tokio::signal::unix::{signal, SignalKind}; + let term_fut = async { + match signal(SignalKind::terminate()) { + Ok(mut s) => { s.recv().await; } + Err(_) => std::future::pending::<()>().await, + } + }; + let int_fut = async { + match signal(SignalKind::interrupt()) { + Ok(mut s) => { s.recv().await; } + Err(_) => std::future::pending::<()>().await, + } + }; + tokio::select! { _ = term_fut => {}, _ = int_fut => {} } +} diff --git a/server/src/projects.rs b/server/src/projects.rs new file mode 100644 index 0000000..f5305da --- /dev/null +++ b/server/src/projects.rs @@ -0,0 +1,219 @@ +use axum::extract::Path; +use axum::Json; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::auth::{audit, AuthUser}; +use crate::error::{AppError, AppResult}; +use crate::util::{new_id, now_iso, sha256_hex, unified_diff}; + +pub type Cx = crate::state::Cx; + +#[derive(Serialize)] +pub struct Project { + pub id: String, + pub name: String, + pub description: String, + pub draft_code: String, + pub draft_generation: i64, + pub created_at: String, + pub updated_at: String, +} + +pub async fn load_owned(cx: &Cx, user_id: &str, project_id: &str) -> AppResult<Project> { + cx.with_db(|db| { + db.query_row("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1 AND user_id=?2", + rusqlite::params![project_id, user_id], row_project) + .map_err(|_| AppError::not_found("project not found")) + }).await +} + +fn row_project(r: &rusqlite::Row) -> rusqlite::Result<Project> { + Ok(Project { + id: r.get(0)?, name: r.get(1)?, description: r.get(2)?, + draft_code: r.get(3)?, draft_generation: r.get(4)?, + created_at: r.get(5)?, updated_at: r.get(6)?, + }) +} + +#[derive(Deserialize)] +pub struct CreateProject { pub name: String, pub description: Option<String> } + +pub async fn create(cx: Cx, auth: AuthUser, body: Option<Json<CreateProject>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let name = r.name.trim().to_string(); + if name.is_empty() || name.len() > 200 { return Err(AppError::bad("validation", "name required (max 200 chars)")); } + let id = new_id(); + let desc = r.description.unwrap_or_default(); + let ts = now_iso(); + let uid = auth.id.clone(); + cx.with_db(|db| -> AppResult<()> { + db.execute("INSERT INTO projects (id,user_id,name,description,draft_code,draft_generation,created_at,updated_at) VALUES (?1,?2,?3,?4,'',0,?5,?5)", + rusqlite::params![&id, &uid, &name, &desc, &ts])?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "project_create", &id, "ok").await; + let p = load_owned(&cx, &auth.id, &id).await?; + Ok((axum::http::StatusCode::CREATED, Json(json!(p)))) +} + +pub async fn list(cx: Cx, auth: AuthUser) -> AppResult<Json<Value>> { + let items: Vec<Project> = cx.with_db(|db| { + let mut st = db.prepare("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE user_id=?1 ORDER BY updated_at DESC")?; + let mut rows = st.query([auth.id.clone()])?; + let mut out = Vec::new(); + while let Some(r) = rows.next()? { out.push(row_project(r)?); } + Ok::<_, AppError>(out) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + let p = load_owned(&cx, &auth.id, &id).await?; + Ok(Json(json!(p))) +} + +#[derive(Deserialize)] +pub struct PatchProject { pub name: Option<String>, pub description: Option<String> } + +pub async fn patch(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<PatchProject>>) -> AppResult<Json<Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if let Some(n) = &r.name { + if n.trim().is_empty() { return Err(AppError::bad("validation", "name must not be empty")); } + } + load_owned(&cx, &auth.id, &id).await?; + cx.with_db(|db| -> AppResult<()> { + db.execute("UPDATE projects SET name=COALESCE(?1,name), description=COALESCE(?2,description), updated_at=?3 WHERE id=?4", + rusqlite::params![r.name, r.description, now_iso(), &id])?; + Ok(()) + }).await?; + let p = load_owned(&cx, &auth.id, &id).await?; + Ok(Json(json!(p))) +} + +#[derive(Deserialize)] +pub struct PutDraft { pub code: String, pub expected_generation: i64 } + +pub async fn put_draft(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<PutDraft>>) -> AppResult<Json<Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + if r.code.len() > crate::config::MAX_CODE_LEN { return Err(AppError::bad("validation", "code too long")); } + let p: Project = cx.with_db(|db| -> AppResult<Project> { + let cur: i64 = db.query_row("SELECT draft_generation FROM projects WHERE id=?1 AND user_id=?2", + rusqlite::params![&id, &auth.id], |row| row.get(0)) + .map_err(|_| AppError::not_found("project not found"))?; + if cur != r.expected_generation { + return Err(AppError::conflict("stale_generation", "draft changed; reload first") + .with_details(json!({ "current_generation": cur }))); + } + db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4", + rusqlite::params![&r.code, cur + 1, now_iso(), &id])?; + db.query_row("SELECT id,name,description,draft_code,draft_generation,created_at,updated_at FROM projects WHERE id=?1", [&id], row_project) + .map_err(AppError::from) + }).await?; + Ok(Json(json!(p))) +} + +// ---- versions ---- + +#[derive(Serialize)] +pub struct Version { + pub id: String, + pub project_id: String, + pub code: String, + pub hash: String, + pub message: String, + pub created_at: String, + pub source: String, +} + +fn row_version(r: &rusqlite::Row) -> rusqlite::Result<Version> { + Ok(Version { id: r.get(0)?, project_id: r.get(1)?, code: r.get(2)?, hash: r.get(3)?, message: r.get(4)?, created_at: r.get(5)?, source: r.get(6)? }) +} + +fn select_version(db: &mut rusqlite::Connection, vid: &str, pid: &str) -> rusqlite::Result<Version> { + let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE id=?1 AND project_id=?2")?; + st.query_row(rusqlite::params![vid, pid], row_version) +} + +pub async fn list_versions(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + load_owned(&cx, &auth.id, &id).await?; + let mut items: Vec<Version> = cx.with_db(|db| { + let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE project_id=?1 ORDER BY created_at ASC")?; + let mut rows = st.query([id.clone()])?; + let mut out = Vec::new(); + while let Some(r) = rows.next()? { out.push(row_version(r)?); } + Ok::<_, AppError>(out) + }).await?; + items.reverse(); + Ok(Json(json!({ "items": items }))) +} + +pub async fn get_version(cx: Cx, auth: AuthUser, Path((pid, vid)): Path<(String, String)>) -> AppResult<Json<Value>> { + load_owned(&cx, &auth.id, &pid).await?; + let v: Version = cx.with_db(|db| select_version(db, &vid, &pid)) + .await + .map_err(|_| AppError::not_found("version not found"))?; + Ok(Json(json!(v))) +} + +pub async fn diff_versions(cx: Cx, auth: AuthUser, Path((pid, vid)): Path<(String, String)>, body: Option<Json<Value>>) -> AppResult<Json<Value>> { + let base: Value = body.map(|b| b.0).unwrap_or(json!({})); + load_owned(&cx, &auth.id, &pid).await?; + let v: Version = cx.with_db(|db| select_version(db, &vid, &pid)) + .await + .map_err(|_| AppError::not_found("version not found"))?; + let base_code: String = match base.get("base_version").and_then(|b| b.as_str()) { + Some(bv) => cx.with_db(|db| { + db.query_row("SELECT code FROM project_versions WHERE id=?1 AND project_id=?2", rusqlite::params![bv, &pid], |r| r.get(0)) + .map_err(|_| AppError::bad("validation", "base_version not found")) + }).await?, + None => String::new(), + }; + Ok(Json(json!({ "diff": unified_diff(&base_code, &v.code), "version": { "id": v.id, "hash": v.hash, "message": v.message } }))) +} + +pub async fn create_version(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<Value>>) -> AppResult<(axum::http::StatusCode, Json<Value>)> { + let b = body.map(|b| b.0).unwrap_or(json!({})); + let message = b.get("message").and_then(|m| m.as_str()).unwrap_or("").to_string(); + if message.trim().is_empty() { return Err(AppError::bad("validation", "message required")); } + let v = create_version_from_draft(&cx, &auth.id, &id, &message, "manual").await?; + Ok((axum::http::StatusCode::CREATED, Json(json!(v)))) +} + +/// Snapshot the current draft as an immutable version. +pub async fn create_version_from_draft(cx: &Cx, user_id: &str, project_id: &str, message: &str, source: &str) -> AppResult<Version> { + let p = load_owned(cx, user_id, project_id).await?; + if p.draft_code.trim().is_empty() { return Err(AppError::bad("validation", "draft is empty; nothing to version")); } + let hash = sha256_hex(p.draft_code.as_bytes()); + let draft = p.draft_code.clone(); + let v: Version = cx.with_db(|db| -> AppResult<Version> { + let id = new_id(); + let ts = now_iso(); + db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,?5,?6,?7)", + rusqlite::params![&id, project_id, &draft, &hash, message, source, &ts])?; + let mut st = db.prepare("SELECT id,project_id,code,hash,message,created_at,source FROM project_versions WHERE id=?1")?; + st.query_row([&id], row_version).map_err(AppError::from) + }).await?; + audit(cx, Some(user_id), "version_create", project_id, "ok").await; + Ok(v) +} + +#[derive(Deserialize)] +pub struct RestoreReq { pub version_id: String, pub expected_generation: i64 } + +pub async fn restore(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<RestoreReq>>) -> AppResult<Json<Value>> { + let Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + cx.with_db(|db| -> AppResult<()> { + let cur: i64 = db.query_row("SELECT draft_generation FROM projects WHERE id=?1 AND user_id=?2", rusqlite::params![&id, &auth.id], |row| row.get(0)) + .map_err(|_| AppError::not_found("project not found"))?; + if cur != r.expected_generation { return Err(AppError::conflict("stale_generation", "draft changed; reload first")); } + let code: String = db.query_row("SELECT code FROM project_versions WHERE id=?1 AND project_id=?2", rusqlite::params![&r.version_id, &id], |row| row.get(0)) + .map_err(|_| AppError::not_found("version not found"))?; + db.execute("UPDATE projects SET draft_code=?1, draft_generation=?2, updated_at=?3 WHERE id=?4", + rusqlite::params![&code, cur + 1, now_iso(), &id])?; + Ok(()) + }).await?; + audit(&cx, Some(&auth.id), "version_restore", &id, "ok").await; + let p = load_owned(&cx, &auth.id, &id).await?; + Ok(Json(json!(p))) +} diff --git a/server/src/runs.rs b/server/src/runs.rs new file mode 100644 index 0000000..8bb934f --- /dev/null +++ b/server/src/runs.rs @@ -0,0 +1,281 @@ +use axum::extract::{Path, Query}; +use axum::http::StatusCode; +use axum::Json; +use serde_json::{json, Value}; + +use crate::auth::{audit, AuthUser}; +use crate::config::MAX_CODE_LEN; +use crate::error::{AppError, AppResult}; +use crate::util::new_id; +use crate::util::now_iso; + +pub use crate::state::Cx; + +const RUN_COLS: &str = "id,project_id,version_id,config,dataset_id,status,error,manifest_hash,created_at,result,started_at,finished_at"; + +fn row_run(r: &rusqlite::Row) -> rusqlite::Result<Value> { + let config: String = r.get(3)?; + let result: Option<String> = r.get(9)?; + Ok(json!({ + "id": r.get::<_, String>(0)?, + "project_id": r.get::<_, String>(1)?, + "version_id": r.get::<_, String>(2)?, + "config": serde_json::from_str::<Value>(&config).unwrap_or(Value::Null), + "dataset_id": r.get::<_, String>(4)?, + "status": r.get::<_, String>(5)?, + "error": r.get::<_, Option<String>>(6)?, + "manifest_hash": r.get::<_, Option<String>>(7)?, + "data_manifest_hash": r.get::<_, Option<String>>(7)?, + "created_at": r.get::<_, String>(8)?, + "result": result.and_then(|s| serde_json::from_str::<Value>(&s).ok()).unwrap_or(Value::Null), + "started_at": r.get::<_, Option<String>>(10)?, + "finished_at": r.get::<_, Option<String>>(11)?, + })) +} + +fn select_run(db: &mut rusqlite::Connection, run_id: &str) -> AppResult<Value> { + let cols = RUN_COLS; + db.query_row(format!("SELECT {cols} FROM runs WHERE id=?1").as_str(), [run_id], row_run) + .map_err(|_| AppError::not_found("run not found")) +} + +async fn assert_run_owner(cx: &Cx, user_id: &str, run_id: &str) -> AppResult<()> { + let owner: Option<String> = cx.with_db(|db| { + db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get(0)).ok() + }).await; + match owner { + Some(u) if u == user_id => Ok(()), + _ => Err(AppError::not_found("run not found")), + } +} + +pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + assert_run_owner(&cx, &auth.id, &id).await?; + let v = cx.with_db(|db| select_run(db, &id)).await?; + Ok(Json(v)) +} + +pub async fn list(cx: Cx, auth: AuthUser, q: Query<std::collections::HashMap<String, String>>) -> AppResult<Json<Value>> { + let project = q.get("project_id").cloned(); + let cols = RUN_COLS; + let items: Vec<Value> = cx.with_db(|db| -> AppResult<Vec<Value>> { + let owned = project.as_deref(); + let sql = if owned.is_some() { + format!("SELECT {cols} FROM runs WHERE user_id=?1 AND project_id=?2 ORDER BY created_at DESC LIMIT 200") + } else { + format!("SELECT {cols} FROM runs WHERE user_id=?1 ORDER BY created_at DESC LIMIT 200") + }; + let mut st = db.prepare(&sql)?; + let mut rows = if let Some(p) = owned { + st.query(rusqlite::params![auth.id, p])? + } else { + st.query([auth.id.clone()])? + }; + let mut out = Vec::new(); + while let Some(r) = rows.next()? { out.push(row_run(r)?); } + Ok(out) + }).await?; + Ok(Json(json!({ "items": items }))) +} + +fn build_config(j: &Value) -> AppResult<Value> { + let g = |k: &str, d: f64| j.get(k).and_then(|v| v.as_f64()).unwrap_or(d); + let capital = j.get("capital").and_then(|v| v.as_f64()); + let config = json!({ + "capital": capital.unwrap_or(1_000_000.0), + "commission": g("commission", 0.0003), + "slippage": g("slippage", 0.001), + "benchmark_symbol": j.get("benchmark_symbol").cloned().unwrap_or(Value::Null), + "parameters": j.get("parameters").cloned().unwrap_or(json!({})), + "seed": j.get("seed").cloned().unwrap_or(Value::Null), + }); + let cap = config.get("capital").and_then(|v| v.as_f64()).unwrap(); + if !cap.is_finite() || !(1.0..=1e12).contains(&cap) { + return Err(AppError::bad("validation", "capital must be between 1 and 1e12")); + } + for k in ["commission", "slippage"] { + let v = config.get(k).and_then(|v| v.as_f64()).unwrap(); + if !v.is_finite() || !(0.0..=0.05).contains(&v) { + return Err(AppError::bad("validation", format!("{k} must be between 0 and 0.05"))); + } + } + if config.get("parameters").and_then(|p| p.as_object()).map(|o| !o.is_empty()).unwrap_or(false) { + if serde_json::to_string(config.get("parameters").unwrap()).unwrap_or_default().len() > MAX_CODE_LEN { + return Err(AppError::bad("validation", "parameters too large")); + } + } + Ok(config) +} + +/// Insert a queued run inside one transaction with an atomic per-user daily quota check. +async fn insert_run(cx: &Cx, auth: &AuthUser, project_id: &str, dataset_id: &str, version_id: &str, config: &Value, manifest_hash: &Option<String>) -> AppResult<String> { + let uid = auth.id.clone(); + let cfg_s = config.to_string(); + let limit = auth.daily_run_limit; + let today: String = chrono::Utc::now().format("%Y-%m-%d").to_string(); + let pattern = format!("{today}%"); + let id = new_id(); + let now = now_iso(); + cx.with_db(|db| -> AppResult<String> { + db.execute("BEGIN IMMEDIATE", []).ok(); + let used: i64 = db.query_row( + "SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2", + rusqlite::params![&uid, &pattern], |r| r.get(0)).unwrap_or(0); + if used >= limit { + db.execute("ROLLBACK", []).ok(); + return Err(AppError::new(axum::http::StatusCode::TOO_MANY_REQUESTS, "run_limit", "daily run quota reached")); + } + db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,manifest_hash,created_at) VALUES (?1,?2,?3,?4,?5,'queued',?6,?7,?8)", + rusqlite::params![&id, &uid, project_id, version_id, dataset_id, &cfg_s, manifest_hash, &now])?; + db.execute("COMMIT", []).map_err(AppError::from)?; + Ok(id) + }).await +} + +pub async fn enqueue(cx: Cx, auth: AuthUser, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> { + let Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?; + let project_id = j.get("project_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "project_id required"))?.to_string(); + let dataset_id = j.get("dataset_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "dataset_id required"))?.to_string(); + + // dataset must exist, belong to the caller and be ready + let (ready, manifest_hash): (bool, Option<String>) = cx.with_db(|db| { + db.query_row("SELECT status='ready', manifest_hash FROM datasets WHERE id=?1 AND user_id=?2", + rusqlite::params![&dataset_id, &auth.id], |r| Ok((r.get::<_, i64>(0)? != 0, r.get(1)?))) + .map_err(|_| AppError::not_found("dataset not found")) + }).await?; + if !ready { return Err(AppError::conflict("dataset_not_ready", "dataset pending or failed; cannot run")); } + + // snapshot + pin draft + let draft: String = cx.with_db(|db| { + db.query_row("SELECT draft_code FROM projects WHERE id=?1 AND user_id=?2", rusqlite::params![&project_id, &auth.id], |r| r.get(0)) + .map_err(|_| AppError::not_found("project not found")) + }).await?; + if draft.trim().is_empty() { + return Err(AppError::bad("validation", "draft is empty; cannot run empty strategy")); + } + + let config = build_config(&j)?; + let version_id = push_run_version(&cx, &auth.id, &project_id, &draft).await?; + + // warnings need explicit acknowledgement + let warnings: Value = cx.with_db(|db| { + let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default(); + let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null); + Ok::<_, AppError>(mj.get("warnings").cloned().unwrap_or(json!([]))) + }).await?; + let mut warnings = warnings; + if let Some(objs) = warnings.as_array_mut() { + // explicit unsupported-restriction surfacing for index feeds + let has_index: bool = cx.with_db(|db| { + let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default(); + let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null); + Ok::<_, AppError>(mj.get("objects").and_then(|o| o.as_array()).map(|a| { + a.iter().any(|o| o.get("instrument").and_then(|i| i.get("asset_type")).and_then(|t| t.as_str()) == Some("index")) + }).unwrap_or(false)) + }).await?; + if has_index && !objs.iter().any(|w| w.as_str().map(|s| s.contains("index feeds are nontradable")).unwrap_or(false)) { + objs.push(json!("index feeds are nontradable research proxies; direct index orders are rejected by the engine")); + } + } + let has_warns = warnings.as_array().map(|a| !a.is_empty()).unwrap_or(false); + if has_warns && !j.get("acknowledge_warnings").and_then(|v| v.as_bool()).unwrap_or(false) { + return Err(AppError::conflict("warnings_unacknowledged", "dataset has warnings; acknowledge to run") + .with_details(json!({"warnings": warnings}))); + } + + let run_id = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &config, &manifest_hash).await?; + audit(&cx, Some(&auth.id), "run_create", &run_id, "ok").await; + let v = cx.with_db(|db| select_run(db, &run_id)).await?; + Ok((StatusCode::ACCEPTED, Json(v))) +} + +/// Snapshot the draft as a pinned run version, reusing an identical hash version. +async fn push_run_version(cx: &Cx, _user_id: &str, project_id: &str, code: &str) -> AppResult<String> { + let hash = crate::util::sha256_hex(code.as_bytes()); + let code = code.to_string(); + cx.with_db(|db| -> AppResult<String> { + let vid: Option<String> = db.query_row("SELECT id FROM project_versions WHERE project_id=?1 AND hash=?2", + rusqlite::params![project_id, &hash], |r| r.get(0)).ok(); + if let Some(v) = vid { return Ok(v); } + let id = new_id(); + db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,'run snapshot','run',?5)", + rusqlite::params![&id, project_id, &code, &hash, now_iso()])?; + Ok(id) + }).await +} + +pub async fn cancel(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> { + assert_run_owner(&cx, &auth.id, &id).await?; + let status: Option<String> = cx.with_db(|db| { + db.query_row("SELECT status FROM runs WHERE id=?1", [&id], |r| r.get(0)).ok() + }).await; + if matches!(status.as_deref(), Some("queued") | Some("running")) { + let res = crate::jobs::signal_cancel(&cx, &id).await; + audit(&cx, Some(&auth.id), "run_cancel", &id, if res { "ok" } else { "not_running" }).await; + } + let v = cx.with_db(|db| select_run(db, &id)).await?; + Ok(Json(v)) +} + +/// Fresh run pinned to the original code+config+dataset; never refetches data. +pub async fn rerun(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> { + assert_run_owner(&cx, &auth.id, &id).await?; + let j = body.map(|b| b.0).unwrap_or(json!({})); + if j.get("use_original_data").and_then(|v| v.as_bool()) != Some(true) { + return Err(AppError::bad("validation", "use_original_data must be true; rerun is pinned to original data")); + } + let (project_id, version_id, dataset_id, config, manifest_hash): (String, String, String, String, Option<String>) = cx.with_db(|db| { + db.query_row("SELECT project_id,version_id,dataset_id,config,manifest_hash FROM runs WHERE id=?1", [&id], |r| + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?, r.get::<_, String>(3)?, r.get::<_, Option<String>>(4)?))) + .map_err(|_| AppError::not_found("run not found")) + }).await?; + // dataset must still be ready and owned + let ready: bool = cx.with_db(|db| { + db.query_row("SELECT status='ready' FROM datasets WHERE id=?1 AND user_id=?2", + rusqlite::params![&dataset_id, &auth.id], |r| r.get(0)) + .map_err(|_| AppError::not_found("dataset not found")) + }).await?; + if !ready { return Err(AppError::conflict("dataset_not_ready", "original dataset is missing or not ready")); } + let cfg_val: Value = serde_json::from_str(&config).unwrap_or(json!({})); + let new_id_r = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &cfg_val, &manifest_hash).await?; + audit(&cx, Some(&auth.id), "run_rerun", &new_id_r, "ok").await; + let v = cx.with_db(|db| select_run(db, &new_id_r)).await?; + Ok((StatusCode::ACCEPTED, Json(v))) +} + +/// Mark running runs failed after a server restart (queued stays resumable). +pub async fn cleanup_interrupted(cx: &Cx) { + cx.with_db(|db| { + db.execute("UPDATE runs SET status='failed', error='interrupted by server restart', finished_at=?1 WHERE status='running'", + rusqlite::params![now_iso()]).ok(); + }).await; +} + + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn config_bounds_enforced() { + let ok = json!({"capital": 100000, "commission": 0.0003, "slippage": 0.001}); + assert!(build_config(&ok).is_ok()); + let bad = json!({"capital": 0.0}); + assert_eq!(build_config(&bad).unwrap_err().code, "validation"); + let bad = json!({"capital": 5000, "commission": 0.5}); + assert_eq!(build_config(&bad).unwrap_err().code, "validation"); + let big = json!({"capital": 100000, "parameters": {"huge": "x".repeat(MAX_CODE_LEN + 1)}}); + assert_eq!(build_config(&big).unwrap_err().code, "validation"); + let defaults = build_config(&json!({})).unwrap(); + assert_eq!(defaults["commission"], json!(0.0003), "defaults are visible to users"); + } + + #[test] + fn nonfinite_sanitized() { + let mut v = json!({"metrics": {"sharpe": f64::NAN, "total_return": 1.0}, "equity": [1.0, f64::INFINITY]}); + crate::jobs::sanitize_nonfinite(&mut v); + let s = serde_json::to_string(&v).unwrap(); + assert!(!s.contains("NaN") && !s.contains("Infinity"), "results must never carry NaN/Inf"); + assert_eq!(v["metrics"]["sharpe"], Value::Null); + } +} diff --git a/server/src/state.rs b/server/src/state.rs new file mode 100644 index 0000000..512bcb3 --- /dev/null +++ b/server/src/state.rs @@ -0,0 +1,55 @@ +use std::sync::Arc; +use tokio::sync::{Mutex as AsyncMutex, Semaphore}; + +use crate::config::Config; + +/// Shared application state. The single SQLite connection is serialized behind +/// an async mutex; concurrent fetch/backtest work is bounded by semaphores. +pub struct AppState { + pub cfg: Config, + pub db: AsyncMutex<rusqlite::Connection>, + pub run_sem: Arc<Semaphore>, + pub fetch_sem: Arc<Semaphore>, +} + +/// Fixed shared contract alias for handler arguments: the axum State extractor +/// over `Arc<AppState>`. Handlers take `cx: Cx`, helpers take `&Cx`. +pub type Cx = axum::extract::State<std::sync::Arc<AppState>>; + +/// Legacy alias kept for modules (auth, projects, datasets) whose handlers take +/// the bare `Arc<AppState>`; both styles are valid extractors for this state. +pub type S = Arc<AppState>; + +impl AppState { + /// Lock the connection, run the closure, return its result exactly. + /// Never `.await` inside the closure; keep transactions in ONE closure. + pub async fn with_db<R>(&self, f: impl FnOnce(&mut rusqlite::Connection) -> R) -> R { + let mut db = self.db.lock().await; + f(&mut db) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + + #[tokio::test] + async fn with_db_returns_closure_result_and_composes() { + let cfg = Config::from_env(); + let conn = rusqlite::Connection::open_in_memory().unwrap(); + let st = AppState { + cfg: cfg.clone(), + db: AsyncMutex::new(conn), + run_sem: Arc::new(Semaphore::new(1)), + fetch_sem: Arc::new(Semaphore::new(1)), + }; + let n: i64 = st.with_db(|db| { + db.execute("CREATE TABLE t(x INTEGER)", []).unwrap(); + db.execute("INSERT INTO t VALUES (42)", []).unwrap(); + db.query_row("SELECT SUM(x) FROM t", [], |r| r.get(0)).unwrap() + }).await; + assert_eq!(n, 42); + st.with_db(|_db| ()).await; + } +} diff --git a/server/src/store.rs b/server/src/store.rs new file mode 100644 index 0000000..f371d10 --- /dev/null +++ b/server/src/store.rs @@ -0,0 +1,206 @@ +use std::path::{Path, PathBuf}; + +use sha2::{Digest, Sha256}; + +/// Immutable content-addressed object storage under {data_dir}/objects/{hash[:2]}/{hash}.{ext}. +/// Content is deduplicated: storing identical bytes twice keeps the first object. +pub struct ObjectStore { + pub root: PathBuf, +} + +/// One ingested artifact (file content hashed and copied into the object store). +#[derive(Debug, Clone)] +pub struct StoredObject { + /// sha256 of content + pub hash: String, + /// path relative to the object root (immutable stored path) + pub stored_path: String, + /// file name as produced by the worker, safe for /data mounts + pub mount_name: String, + pub size: u64, +} + +impl ObjectStore { + pub fn new(data_dir: &str) -> Self { + ObjectStore { root: Path::new(data_dir).join("objects") } + } + + /// Store raw bytes by content hash. Returns (hash, stored relative path). + pub fn store(&self, bytes: &[u8], filename: &str) -> std::io::Result<(String, String)> { + let hash = format!("{:x}", Sha256::digest(bytes)); + let dir = self.root.join(&hash[..2]); + std::fs::create_dir_all(&dir)?; + let ext = safe_ext(filename); + let path = dir.join(format!("{hash}.{ext}")); + if !path.is_file() { + // Atomic write in the final directory; suffix append (not with_extension) + // so different source extensions cannot collide on one tmp name. + let tmp = dir.join(format!("{hash}.{ext}.tmp")); + std::fs::write(&tmp, bytes)?; + std::fs::rename(&tmp, &path)?; + } + let rel = path + .strip_prefix(&self.root) + .map(|p| p.to_string_lossy().into_owned()) + .unwrap_or_else(|_| path.to_string_lossy().into_owned()); + Ok((hash, rel)) + } + + /// Map a stored relative path (server controlled) back to an absolute path. + pub fn absolute(&self, rel: &str) -> PathBuf { + self.root.join(rel) + } + + /// Hash and ingest every regular file under an output directory (worker + /// artifacts). Returns one entry per file, sorted for determinism. + /// Rejects symlinked entries. + pub fn ingest_directory(&self, dir: &Path) -> std::io::Result<Vec<StoredObject>> { + let mut files: Vec<PathBuf> = Vec::new(); + collect_files(dir, dir, &mut files)?; + files.sort(); + let mut out = Vec::with_capacity(files.len()); + for f in files { + let is_symlink = f.symlink_metadata()?.file_type().is_symlink() + || std::fs::symlink_metadata(&f)?.file_type().is_symlink(); + if is_symlink { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "symlinked artifact rejected", + )); + } + let bytes = std::fs::read(&f)?; + let mount_name = f + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("artifact.bin") + .to_string(); + let rel_path = f.strip_prefix(dir).expect("strip_prefix"); + let (hash, stored_path) = + self.store(&bytes, &mount_name)?; + out.push(StoredObject { + hash, + stored_path, + mount_name: rel_path.display().to_string(), + size: bytes.len() as u64, + }); + } + Ok(out) + } +} + +fn collect_files(_root: &Path, dir: &Path, out: &mut Vec<PathBuf>) -> std::io::Result<()> { + for entry in std::fs::read_dir(dir)? { + let p = entry?.path(); + let ty = p.symlink_metadata()?.file_type(); + if ty.is_symlink() { + // path traversal defense: no symlinked artifacts, ever + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "symlink in artifact tree rejected", + )); + } + if ty.is_dir() { + collect_files(_root, &p, out)?; + } else { + out.push(p); + } + } + Ok(()) +} + +/// Sanitize a filename for use as an object extension: only the final +/// extension survives, restricted to alphanumeric chars. +fn safe_ext(filename: &str) -> String { + let base = filename.rsplit('/').next().unwrap_or("data"); + let e = base.rsplit('.').next().unwrap_or("bin").to_string(); + let v: String = e.chars().filter(|c| c.is_ascii_alphanumeric()).collect(); + if v.is_empty() || v.parse::<usize>().is_ok() { + "bin".into() + } else { + v.to_lowercase() + } +} + +/// Public wrapper used when a filename has no usable extension. +pub fn filename_or_bin(name: &str, fallback: &str) -> String { + let base = name.rsplit('/').next().unwrap_or(fallback); + let v: String = base + .chars() + .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_')) + .collect(); + if v.is_empty() || v == "." { + fallback.into() + } else { + v + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_store(tag: &str) -> (tempfile::TempDir, ObjectStore) { + let t = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let store = ObjectStore::new(t.path().join(tag).to_str().unwrap()); + (t, store) + } + + #[test] + fn store_is_content_addressed_and_deduplicated() { + let (_t, s) = tmp_store("objs1"); + let (h1, p1) = s.store(b"hello world", "a.csv").unwrap(); + let (h2, p2) = s.store(b"hello world", "b.csv").unwrap(); + assert_eq!(h1, h2); + assert_eq!(p1, p2); + assert!(p1.starts_with(&h1[..2]), "{p1}"); + let abs = s.absolute(&p1); + assert_eq!(std::fs::read(&abs).unwrap(), b"hello world".to_vec()); + assert!(s.root.join(&p1) == abs, "stored path resolves under root"); + } + + #[test] + fn same_stem_different_extension_no_collision() { + let (_t, s) = tmp_store("objs2"); + let (_, a) = s.store(b"csv-bytes", "obj.csv").unwrap(); + let (_, b) = s.store(b"json-bytes", "obj.json").unwrap(); + assert_ne!(a, b); + assert!(!a.ends_with(".tmp")); + assert!(std::fs::read(s.absolute(&a)).unwrap().starts_with(b"csv")); + } + + #[test] + fn ingest_directory_walks_and_rejects_symlinks() { + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let out = td.path().join("out"); + let objd = out.join("objects"); + std::fs::create_dir_all(&objd).unwrap(); + std::fs::write(objd.join("data.csv"), b"date,close\n2024-01-02,10\n").unwrap(); + std::fs::write(out.join("result.json"), b"{\"status\":\"ready\"}").unwrap(); + let (_t, s) = tmp_store("objs3"); + let stored = s.ingest_directory(&out).unwrap(); + assert_eq!(stored.len(), 2); + let names: Vec<&str> = stored.iter().map(|o| o.mount_name.as_str()).collect(); + assert!(names.contains_all(&["objects/data.csv", "result.json"]), "{names:?}"); + // same content re-ingested maps to the same stored object + let again = s.ingest_directory(&out).unwrap(); + for o in &stored { + assert!(again.iter().any(|n| n.hash == o.hash)); + } + // symlink rejection + std::os::unix::fs::symlink( + objd.join("data.csv"), + objd.join("data_link.csv"), + ) + .unwrap(); + assert!(s.ingest_directory(&out).is_err()); + } +} + +trait ContainsAll { + fn contains_all(&self, needles: &[&str]) -> bool; +} +impl ContainsAll for Vec<&str> { + fn contains_all(&self, needles: &[&str]) -> bool { + needles.iter().all(|n| self.iter().any(|m| m.contains(n))) + } +} diff --git a/server/src/util.rs b/server/src/util.rs new file mode 100644 index 0000000..5a82806 --- /dev/null +++ b/server/src/util.rs @@ -0,0 +1,138 @@ +use sha2::{Digest, Sha256}; + +pub fn sha256_hex(data: &[u8]) -> String { + let mut h = Sha256::new(); + h.update(data); + hex::encode(h.finalize()) +} + +/// Synchronous UTC timestamp helper, ISO 8601 with millisecond precision. +pub fn now_iso() -> String { + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true) +} + +pub fn plus_hours(hours: i64) -> String { + (chrono::Utc::now() + chrono::Duration::hours(hours)) + .to_rfc3339_opts(chrono::SecondsFormat::Millis, true) +} + +pub fn new_id() -> String { + uuid::Uuid::new_v4().to_string() +} + +pub fn gen_token() -> String { + use rand::RngCore; + let mut buf = [0u8; 32]; + rand::rngs::OsRng.fill_bytes(&mut buf); + hex::encode(buf) +} + +/// Line based unified diff (full document, no hunks) used by the AI draft diff. +pub fn unified_diff(a: &str, b: &str) -> String { + let x: Vec<&str> = a.lines().collect(); + let y: Vec<&str> = b.lines().collect(); + let n = x.len(); + let m = y.len(); + // LCS table + let mut dp = vec![vec![0usize; m + 1]; n + 1]; + for i in (0..n).rev() { + for j in (0..m).rev() { + dp[i][j] = if x[i] == y[j] { + dp[i + 1][j + 1] + 1 + } else { + dp[i + 1][j].max(dp[i][j + 1]) + }; + } + } + let mut out = String::new(); + let mut ctx = std::collections::VecDeque::<usize>::new(); + let (mut i, mut j) = (0usize, 0usize); + while i < n || j < m { + if i < n && j < m && x[i] == y[j] { + ctx.push_back(i); + if ctx.len() > 2 { + let keep = ctx.pop_front().unwrap(); + out.push_str(&format!(" {}\n", x[keep])); + } + i += 1; + j += 1; + } else { + // entering a change: emit queued context lines first + for k in ctx.drain(..) { + out.push_str(&format!(" {}\n", x[k])); + } + if i < n && (j >= m || dp[i + 1][j] >= dp[i][j + 1]) { + out.push_str(&format!("-{}\n", x[i])); + i += 1; + } else { + out.push_str(&format!("+{}\n", y[j])); + j += 1; + } + } + } + out +} + +/// Deterministic canonical JSON string (recursively sorted keys) for hashing. +pub fn canonical_json(v: &serde_json::Value) -> String { + fn sort(v: &serde_json::Value) -> serde_json::Value { + match v { + serde_json::Value::Array(a) => serde_json::Value::Array(a.iter().map(sort).collect()), + serde_json::Value::Object(o) => { + let mut keys: Vec<(String, serde_json::Value)> = + o.iter().map(|(k, v)| (k.clone(), sort(v))).collect(); + keys.sort_by(|a, b| a.0.cmp(&b.0)); + serde_json::Value::Object(keys.into_iter().collect()) + } + other => other.clone(), + } + } + sort(v).to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn now_iso_is_synchronous_iso8601() { + let t = now_iso(); + let parsed = chrono::DateTime::parse_from_rfc3339(&t).expect("ISO timestamp"); + assert!(parsed.timestamp() > 1_700_000_000); + assert!(t.ends_with('Z')); + } + + #[test] + fn plus_hours_and_tokens() { + let e = chrono::DateTime::parse_from_rfc3339(&plus_hours(1)).unwrap(); + let n = chrono::DateTime::parse_from_rfc3339(&now_iso()).unwrap(); + assert!((e - n).num_minutes() >= 59); + let a = gen_token(); + assert_eq!(a.len(), 64); + assert_ne!(a, gen_token()); + assert_eq!(new_id().len(), 36); + } + + #[test] + fn sha256_is_stable() { + assert_eq!(sha256_hex(b"abc"), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"); + } + + #[test] + fn unified_diff_marks_changes_with_context() { + let a = "x\nkeep\ny"; + let b = "x\nkeep\nz"; + let d = unified_diff(a, b); + assert!(d.contains("-y\n+z\n"), "diff: {d}"); + assert!(d.contains(" keep")); + let same = unified_diff(a, "x\nkeep\ny"); + assert!(!same.contains('+') && !same.contains('-'), "no diff: {same}"); + } + + #[test] + fn canonical_json_sorts_recursively() { + let v: serde_json::Value = + serde_json::from_str(r#"{"b":1,"a":{"y":[3,2],"x":true}}"#).unwrap(); + assert_eq!(canonical_json(&v), r#"{"a":{"x":true,"y":[3,2]},"b":1}"#); + } +} diff --git a/server/src/worker.rs b/server/src/worker.rs new file mode 100644 index 0000000..8e28de5 --- /dev/null +++ b/server/src/worker.rs @@ -0,0 +1,482 @@ +use std::sync::Arc; + +use std::os::unix::fs::PermissionsExt; + +use tokio::process::Command; + +use crate::error::{AppError, AppResult}; +use crate::state::AppState; + +const NONROOT_UID: u64 = 65534; +const MAX_STDOUT: usize = 64_000; +const MAX_STDERR: usize = 16_000; + +#[derive(Debug)] +pub struct ContainerResult { + pub status: Option<i32>, + pub stdout: String, + pub stderr: String, +} + +impl ContainerResult { + pub fn ok(&self) -> bool { + self.status == Some(0) + } +} + +/// Docker run arguments (after `docker`). Isolation contract: +/// no caps, no privilege escalation, non-root uid 65534, read-only root +/// filesystem, small tmpfs, pids/memory/cpu limits, network only when the +/// task requires the data provider; backtest always runs with --network=none. +pub fn docker_args( + network: bool, + mounts: &[(String, String, bool)], + image: &str, + name: &str, + args: &[String], +) -> Vec<String> { + let mut a: Vec<String> = [ + "run", + "--rm", + // Signals reaching the runner CLI must never be proxied into the + // container: on service stop the container is expected to die via the + // app's exact-name cleanup (cancel/timeout/startup), not via a CLI + // relay; a proxying CLI was observed lingering under systemd + // final-sigterm (TimeoutStopSec exhaustion). + "--sig-proxy=false", + "--name", + name, + "--cap-drop=ALL", + "--security-opt=no-new-privileges", + CONCAT_USER, + "--read-only", + "--tmpfs=/tmp:rw,size=256m,mode=1777", + "--pids-limit=128", + "--memory=2g", + "--cpus=2", + ] + .iter() + .map(|s| s.to_string()) + .collect(); + // explicit network choice; only data fetch/search uses the bridge + a.push(if network { "--network=bridge".into() } else { "--network=none".into() }); + for (src, dst, ro) in mounts { + a.push("-v".into()); + a.push(format!("{}:{}{}", src, dst, if *ro { ":ro" } else { "" })); + } + a.push(image.to_string()); + a.extend_from_slice(args); + a +} + +async fn drain_bounded<R>(rd: R, max: usize) -> String +where + R: tokio::io::AsyncRead + Unpin, +{ + use tokio::io::AsyncReadExt; + let mut buf = Vec::with_capacity(1024); + let mut chunk = [0u8; 8192]; + let mut reader = rd; + loop { + match reader.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => { + // drain everything, but keep only the tail-relevant bounded prefix + if buf.len() < max { + let take = n.min(max - buf.len()); + buf.extend_from_slice(&chunk[..take]); + } + if buf.len() >= max { + // continue draining the pipe without buffering the rest + let mut sink = [0u8; 8192]; + loop { + match reader.read(&mut sink).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + break; + } + } + Err(_) => break, + } + } + truncate(&String::from_utf8_lossy(&buf), max) +} + +async fn execute_docker(full: &[String], name: &str, timeout_secs: u64) -> AppResult<ContainerResult> { + execute_docker_named("docker", full, name, timeout_secs).await +} + +async fn execute_docker_named( + docker_bin: &str, + full: &[String], + name: &str, + timeout_secs: u64, +) -> AppResult<ContainerResult> { + let mut child = Command::new(docker_bin) + .args(full) + // Kill the runner process when the future that owns it is dropped. A + // graceful server stop drops the jobs task; without this the docker + // CLI child would linger inside the systemd cgroup and block the + // unit stop until TimeoutStopSec forced SIGKILL. Bounded lifetime. + .kill_on_drop(true) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .map_err(|e| AppError::internal(format!("failed to spawn worker container: {e}")))?; + let stdout = child.stdout.take().expect("stdout piped"); + let stderr = child.stderr.take().expect("stderr piped"); + let stdout_task = tokio::spawn(drain_bounded(stdout, MAX_STDOUT)); + let stderr_task = tokio::spawn(drain_bounded(stderr, MAX_STDERR)); + + let wait_res = tokio::time::timeout( + std::time::Duration::from_secs(timeout_secs), + child.wait(), + ) + .await; + + let status = match wait_res { + Ok(Ok(st)) => st.code(), + Ok(Err(e)) => { + return Err(AppError::internal(format!("worker process error: {e}")).with_code("runner_failed")) + } + Err(_) => { + // Timeout: kill the specific container by name so user code cannot + // linger; then reap the docker client process. + kill_container(name).await; + let _ = child.wait().await; + return Err(AppError::internal(format!( + "worker container timed out after {timeout_secs}s and was killed: {name}" + )) + .with_code("runner_timeout")); + } + }; + + Ok(ContainerResult { + status, + stdout: stdout_task.await.unwrap_or_default(), + stderr: stderr_task.await.unwrap_or_default(), + }) +} + +/// Kill and remove the named container. Returns true when docker succeeded. +pub async fn cancel_container(name: &str) -> bool { + kill_container(name).await +} + +async fn kill_container(name: &str) -> bool { + Command::new("docker") + .args(["kill", name]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .output() + .await + .ok(); + Command::new("docker") + .args(["rm", "-f", name]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .output() + .await + .map(|o| o.status.success()) + .unwrap_or(false) +} + +pub async fn docker_available() -> bool { + tokio::process::Command::new("docker") + .args(["version", "--format", "ok"]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .output() + .await + .map(|o| o.status.success()) + .unwrap_or(false) +} + +/// Mounts need world permissions: the container runs as uid 65534 while host +/// ownership is the server user. Best effort only. +fn prepare_mounts(mounts: &[(String, String, bool)]) { + for (src, _dst, ro) in mounts { + let p = std::path::Path::new(src); + if !p.is_dir() { + continue; + } + let mode = if *ro { 0o755 } else { 0o777 }; + let _ = std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode)); + // Files inside ro input dirs must be world readable; output files are + // written by the container with its umask. + if *ro { + if let Ok(rd) = std::fs::read_dir(p) { + for e in rd.flatten() { + let fmode = if e.path().is_file() { + std::fs::Permissions::from_mode(0o644) + } else { + std::fs::Permissions::from_mode(0o755) + }; + let _ = std::fs::set_permissions(e.path(), fmode); + } + } + } + } +} + +/// Run the worker image with a fixed container name so cancel maps to one +/// specific container id (never a global prune). +pub async fn run_named( + cx: &Arc<AppState>, + network: bool, + mounts: &[(String, String, bool)], + args: &[String], + name: &str, + timeout_secs: u64, +) -> AppResult<ContainerResult> { + let cfg = &cx.cfg; + prepare_mounts(mounts); + let full = docker_args(network, mounts, &cfg.worker_image, name, args); + execute_docker(&full, name, timeout_secs).await +} + +/// Instrument catalog search through the worker container (network enabled). +/// Returns the JSON array printed by `worker.main search`; failures are real +/// errors, never an empty fake success. Results are cached briefly per query so +/// repeated keystrokes reuse the actual provider identity (full item payloads). +pub async fn search_instruments( + cx: &Arc<AppState>, + query: &str, + limit: i64, +) -> AppResult<Vec<serde_json::Value>> { + let query = query.trim().to_string(); + if query.is_empty() { + return Ok(Vec::new()); + } + let limit = if (1..=100).contains(&limit) { limit } else { 50 }; + let cache_key = format!("q={query}&limit={limit}"); + if let Some(items) = catalog_cache_get(&cache_key) { + return Ok(items); + } + let args = vec![ + "python".into(), + "-m".into(), + "worker.main".into(), + "search".into(), + "--query".into(), + query, + "--limit".into(), + limit.to_string(), + ]; + // The worker enforces <=4s per HTTP source; the container including startup + // is bounded here. This is the outer bound for the whole search round trip. + let res = run_named(cx, true, &[], &args, &random_name(), 12).await?; + if !res.ok() { + return Err(AppError::bad("search_failed", truncate(&res.stderr, 500))); + } + // The contract is a single JSON object envelope on stdout: + // {"status":"ready"|"failed","items":[...],"error":{...},...} + // A `failed` status is surfaced as an error, never as an empty success. + let envelope: serde_json::Value = serde_json::from_str(res.stdout.trim()) + .map_err(|e| AppError::internal(format!("invalid search envelope: {e}")))?; + let status = envelope + .get("status") + .and_then(|v| v.as_str()) + .ok_or_else(|| AppError::internal("invalid search envelope: missing status"))?; + if status != "ready" { + let err = envelope.get("error").cloned().unwrap_or(serde_json::Value::Null); + let code = err + .get("code") + .and_then(|v| v.as_str()) + .unwrap_or("search_unavailable"); + let message = err + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("instrument search providers unavailable"); + // provider error codes are dynamic; they ride in the message so the + // HTTP layer keeps static error codes + return Err(AppError::bad( + "search_unavailable", + format!("[{code}] {message}"), + )); + } + let items: Vec<serde_json::Value> = serde_json::from_value( + envelope.get("items").cloned().unwrap_or(serde_json::Value::Null), + ) + .map_err(|e| AppError::internal(format!("invalid search envelope items: {e}")))?; + catalog_cache_put(&cache_key, &items); + Ok(items) +} + +/// Small in-process TTL cache for catalog search results (provider identity). +const CATALOG_TTL_SECS: u64 = 300; +const CATALOG_MAX_ENTRIES: usize = 128; + +fn catalog_cache() -> &'static tokio::sync::Mutex<std::collections::HashMap<String, (std::time::Instant, Vec<serde_json::Value>)>> { + static MAP: std::sync::OnceLock<tokio::sync::Mutex<std::collections::HashMap<String, (std::time::Instant, Vec<serde_json::Value>)>>> = + std::sync::OnceLock::new(); + MAP.get_or_init(|| tokio::sync::Mutex::new(std::collections::HashMap::new())) +} + +fn catalog_cache_get(key: &str) -> Option<Vec<serde_json::Value>> { + // Instant checks must not block behind stdio work; try_lock is fine here. + let map = catalog_cache().try_lock().ok()?; + let (at, items) = map.get(key)?; + if at.elapsed() < std::time::Duration::from_secs(CATALOG_TTL_SECS) { + Some(items.clone()) + } else { + None + } +} + +fn catalog_cache_put(key: &str, items: &[serde_json::Value]) { + if let Ok(mut map) = catalog_cache().try_lock() { + if map.len() >= CATALOG_MAX_ENTRIES { + map.clear(); + } + map.insert(key.to_string(), (std::time::Instant::now(), items.to_vec())); + } +} + +fn random_name() -> String { + use rand::RngCore; + let mut buf = [0u8; 8]; + rand::rngs::OsRng.fill_bytes(&mut buf); + format!("sl-run-{}", hex::encode(buf)) +} + +pub fn truncate(s: &str, n: usize) -> String { + s.chars().take(n).collect() +} + +const CONCAT_USER: &str = "--user=65534:65534"; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn run_cli_never_proxies_signals_and_names_the_worker() { + // Cause-establishing regression: systemd restart stall happened because + // the runner CLI swallowed SIGTERM while relaying to the container. + // The CLI (and every other container invokation) must use sig-proxy=false. + for network in [true, false] { + let full = docker_args( + network, + &[("/job/output".into(), "/output".into(), false)], + "strategy-lab-worker:local", + "sl-run-t", + &["python".into()], + ); + let s = full.join(" "); + assert!(s.contains("--sig-proxy=false"), "{s}"); + assert!(s.contains("--rm --sig-proxy=false --name sl-run-t")); + assert!(s.contains(format!("--user={NONROOT_UID}:").as_str())); + } + } + + #[tokio::test] + async fn runner_child_is_reaped_when_the_job_future_is_dropped() { + // Focused cause test: a runner child that ignores SIGTERM must not + // outlive its owning future (systemd restart stall cause). We emulate + // with a stub runner in a UNIQUE tempdir that ignores SIGTERM and then + // `exec`s into sleep so the stub PID IS the sleep process: kill_on_drop + // removes it entirely, leaving no grandchild orphan. Wait supervision + // uses exact PID checks (`/proc/<pid>`), never process-name scans. + let td = tempfile::tempdir_in("/tmp/opencode").unwrap(); + let stub_path = td.path().join("runner-stub.sh"); + let pid_file = td.path().join("stub.pid"); + std::fs::write(&stub_path, format!( + "#!/bin/sh\ntrap '' TERM INT\necho $$ > {}\nexec sleep 500\n", pid_file.display() + )).unwrap(); + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&stub_path, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + let stub = stub_path.clone(); + let task = tokio::spawn(async move { + let _res = execute_docker_named( + stub.to_str().unwrap(), + &["long-running-stub".into()], + "sl-run-stub", + 5, + ).await; + }); + // wait until the stub published its exact PID + let mut stub_pid: Option<i32> = None; + for _ in 0..50 { + if let Ok(s) = std::fs::read_to_string(&pid_file) { + stub_pid = s.trim().parse().ok(); + } + if stub_pid.is_some() { break; } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + let stub_pid: i32 = match stub_pid { + Some(p) => p, + None => panic!("stub never published its PID; test setup broken"), + }; + let proc_dir = format!("/proc/{stub_pid}"); + // sanity: stub alive; and after `exec` it IS the sleep grandchild + assert!(std::path::Path::new(&proc_dir).exists(), "stub pid {stub_pid} must be alive before the drop"); + let cmd = std::fs::read_to_string(format!("{proc_dir}/cmdline")).unwrap_or_default(); + assert!(cmd.contains("sleep"), "exec replace failed; test would leave an orphan: {cmd:?}"); + task.abort(); // drops the execute_docker future mid-flight -> kill_on_drop -> SIGKILL + // exact-PID supervision: gone == /proc/<pid> has vanished + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3); + let mut gone = false; + while std::time::Instant::now() < deadline { + if !std::path::Path::new(&proc_dir).exists() { + gone = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert!(gone, "runner stub pid {stub_pid} was not reaped when its future was dropped"); + // no grandchild either: the exec'd sleep adopted the same PID, then was + // SIGKILLed with the rest; nothing named-scan was used. + let _ = stub_path; // file removed with the unique tempdir at scope end + } + fn mounts() -> Vec<(String, String, bool)> { + vec![ + ("/job/input".into(), "/input".into(), true), + ("/job/output".into(), "/output".into(), false), + ("/data/objects/x.csv".into(), "/data/x.csv".into(), true), + ] + } + + #[test] + fn backtest_container_flags_are_bounded_isolated_nonroot() { + let args = vec!["python".to_string(), "-m".to_string(), "worker.main".to_string()]; + let full = docker_args(false, &mounts(), "strategy-lab-worker:local", "sl-run-x", &args); + let s = full.join(" "); + assert!(s.contains("strategy-lab-worker:local")); + assert!(s.contains("--network=none"), "runner must not use network: {s}"); + assert!(s.contains(format!("--user={NONROOT_UID}:").as_str()), "{s}"); + assert!(s.contains("--cap-drop=ALL")); + assert!(s.contains("--security-opt=no-new-privileges")); + assert!(s.contains("--read-only")); + assert!(s.contains("--pids-limit=128")); + assert!(s.contains("--memory=2g")); + assert!(s.contains("--cpus=2")); + assert!(s.contains("--tmpfs=/tmp:")); + assert!(!s.contains("/var/run/docker.sock"), "no docker socket in worker: {s}"); + // mount directions preserved + assert!(s.contains("/job/input:/input:ro")); + assert!(s.contains("/job/output:/output")); + assert!(s.contains("/data/objects/x.csv:/data/x.csv:ro")); + } + + #[test] + fn fetch_container_has_network_and_same_isolation() { + let full = docker_args(true, &[], "img", "sl-fetch-1", &["python".into()]); + let s = full.join(" "); + assert!(!s.contains("--network=none")); + assert!(s.contains("--cap-drop=ALL") && s.contains(CONCAT_USER)); + assert!(s.contains("--rm --sig-proxy=false --name sl-fetch-1")); + } + + #[test] + fn truncate_is_char_safe() { + let s = "中文内容"; + let t = truncate(s, 4); + assert!(t.chars().count() <= 4); + assert_eq!(truncate("short", 100), "short"); + } +} |
