Compare commits
15
Commits
v1.1.0-alpha.19
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
25fa6a488a | ||
|
|
78dacb7e17 | ||
|
|
c2c8336438 | ||
|
|
63c8ef0b8a | ||
|
|
049113cec8 | ||
|
|
330f95a375 | ||
|
|
e744e62d03 | ||
|
|
b9f47e556c | ||
|
|
e6b489f117 | ||
|
|
73dde99e3a | ||
|
|
ecc8944006 | ||
|
|
dc15e2d04b | ||
|
|
ff1fc25d77 | ||
|
|
6b67326c3b | ||
|
|
d57f8778ac |
@@ -104,12 +104,28 @@ jobs:
|
|||||||
working-directory: artifacts
|
working-directory: artifacts
|
||||||
run: sha256sum * > SHA256SUMS.txt
|
run: sha256sum * > SHA256SUMS.txt
|
||||||
|
|
||||||
|
- name: sign the checksums (RSA-2048/SHA-256)
|
||||||
|
env:
|
||||||
|
KEY: ${{ secrets.SCIMESH_SIGNING_KEY }}
|
||||||
|
working-directory: artifacts
|
||||||
|
run: |
|
||||||
|
if [ -n "$KEY" ]; then
|
||||||
|
printf '%s\n' "$KEY" > /tmp/scimesh-sign-key.pem
|
||||||
|
openssl dgst -sha256 -sign /tmp/scimesh-sign-key.pem \
|
||||||
|
-out SHA256SUMS.txt.sig SHA256SUMS.txt
|
||||||
|
echo "signed SHA256SUMS.txt"
|
||||||
|
else
|
||||||
|
echo "SCIMESH_SIGNING_KEY is not set; releasing without a signature"
|
||||||
|
fi
|
||||||
|
|
||||||
- uses: softprops/action-gh-release@v2
|
- uses: softprops/action-gh-release@v2
|
||||||
with:
|
with:
|
||||||
files: |
|
files: |
|
||||||
artifacts/*
|
artifacts/*
|
||||||
install.sh
|
install.sh
|
||||||
install.ps1
|
install.ps1
|
||||||
|
uninstall.sh
|
||||||
|
uninstall.ps1
|
||||||
# Pre-release tags (e.g. v1.1.0-alpha.1) publish as pre-releases.
|
# Pre-release tags (e.g. v1.1.0-alpha.1) publish as pre-releases.
|
||||||
prerelease: ${{ contains(github.ref_name, '-alpha') }}
|
prerelease: ${{ contains(github.ref_name, '-alpha') }}
|
||||||
generate_release_notes: true
|
generate_release_notes: true
|
||||||
|
|||||||
@@ -38,6 +38,22 @@
|
|||||||
## Plan (предыдущая задача — выполнена)
|
## Plan (предыдущая задача — выполнена)
|
||||||
1–7. Docker E2E пайплайна «install как человек → serve → визард → воркер → джоб» — выполнено, см. Progress ниже.
|
1–7. Docker E2E пайплайна «install как человек → serve → визард → воркер → джоб» — выполнено, см. Progress ниже.
|
||||||
|
|
||||||
|
## Progress (день: конкурентность в агенте — параллелизм через нашу архитектуру)
|
||||||
|
- ✅ По просьбе «реализовать параллелизм через нашу архитектуру»: `WORKER_CONCURRENCY` (поле `concurrency` в конфиге визарда, шаг Machine) — агент регистрируется один раз и ведёт N циклов claim→выполнение→upload под одним worker id. N шардов обрабатываются параллельно на одной машине, используя таски координатора как единицу параллелизма; SDK не менялся.
|
||||||
|
- ✅ Реализация: `Config.Concurrency` + env; `Daemon.RunForever` → register once → N goroutine-циклов (общий счётчик MaxTasks под мьютексом); визард: поле «Concurrent task loops» + конфиг; тест с маркер-скриптом доказывает 3 параллельных исполнения (race-тесты зелёные после мьютекса в fake).
|
||||||
|
- ✅ Измерено на релизном коде в Docker (один воркер, 8 шардов): concurrency=1 → 12s; concurrency=4 → 4s (**3×**). Плюс прежний `similarity-search-parallel` (потоки внутри шарда) композируется с конкурентностью.
|
||||||
|
- ✅ Релиз v1.1.0-alpha.20 (бинарники + wheel); полный гейт: race + lint 0 issues + pytest 213.
|
||||||
|
- ✅ Бинарник worker-agent на машине пользователя обновлён до alpha.20.
|
||||||
|
- GIL-высвобождения в RDKit нет (проверено: `RDKIT_ALLOW_GIL_RELEASE` не помогает), поэтому внутришардовые потоки не ускоряют чистый RDKit-путь — конкурентность задач это и компенсирует.
|
||||||
|
|
||||||
|
## Progress (утро: similarity-search-parallel)
|
||||||
|
- ✅ Новый workload `similarity-search-parallel@1.0.0` (отдельная версия, как просил пользователь): подкласс `SimilaritySearchSDKWorkload` + параллельное ядро `search_parallel/core.py` — fingerprinting+скоринг шарда через `ThreadPoolExecutor` (параметр `threads`, default CPU count); `pool.map` сохраняет порядок строк, поэтому merge идентичен последовательному (`_HeapEntry`) и результат **байт-в-байт** равен `similarity-search` при любом числе потоков.
|
||||||
|
- ✅ Тесты: байт-в-байт vs эталон для threads 1/2/4 с намеренными связями (изомеры, дубликаты), executor-прогон, валидация параметров, регистрация манифеста; 213 pytest зелёные.
|
||||||
|
- ✅ Экспортирован в каталог координатора (5 ворклоадов), Go-тесты зелёные; release v1.1.0-alpha.19 (бинарники + wheel `scimesh-1.1.0a19`).
|
||||||
|
- ✅ Распределённый E2E в Docker: воркер с wheel a19, джоб similarity-search-parallel (threads=4) → completed → результат байт-в-байт = локальному эталону.
|
||||||
|
- ✅ Бинарники на машине пользователя обновлены до alpha.19 (для появления ворклоада в UI нужен рестарт serve + переустановка рантайма воркера).
|
||||||
|
- Примечание: в CPython потоки не ускоряют чистый RDKit-путь (GIL), но структура готова к ядрам, отпускающим GIL (numpy и т.п.); при желании можно добавить процесс-пул как отдельный workload в будущей версии протокола.
|
||||||
|
|
||||||
## Progress (ночная сессия)
|
## Progress (ночная сессия)
|
||||||
- ✅ **П.1 Пустое имя воркера**: `domain.NewWorker` нормализует/отклоняет пустое имя + `TestNewWorkerRejectsBlankName` (починен `fixedTime`→`testNow`).
|
- ✅ **П.1 Пустое имя воркера**: `domain.NewWorker` нормализует/отклоняет пустое имя + `TestNewWorkerRejectsBlankName` (починен `fixedTime`→`testNow`).
|
||||||
- ✅ **П.2 `--check`**: пробует managed venv (если установлен) + реальный пробинг учётки (exchange ключа / claim-пробa) — `CheckAuth` + тесты; на машине пользователя: `✓ auth: credential accepted`, venv python, scimesh installed.
|
- ✅ **П.2 `--check`**: пробует managed venv (если установлен) + реальный пробинг учётки (exchange ключа / claim-пробa) — `CheckAuth` + тесты; на машине пользователя: `✓ auth: credential accepted`, venv python, scimesh installed.
|
||||||
@@ -59,8 +75,18 @@
|
|||||||
- ✅ На машине пользователя: визард alpha.15, правильный токен, venv из wheel, воркер emil-pc online, 15 пустых воркеров вычищены из БД.
|
- ✅ На машине пользователя: визард alpha.15, правильный токен, venv из wheel, воркер emil-pc online, 15 пустых воркеров вычищены из БД.
|
||||||
- ✅ Гейт: race + lint + pytest 208.
|
- ✅ Гейт: race + lint + pytest 208.
|
||||||
|
|
||||||
|
## Progress (день 2: контроль версий + изоляция)
|
||||||
|
- ✅ Визард сверяет установленную версию scimesh с версией бинарника: preflight показывает «installed X, worker needs Y — press Install» и снова предлагает кнопку Install (ensureMatchingScimeshVersion, dev-сборки пропускают). Проверено на машине: a21 vs a22 → красный → Install → зелёный 1.1.0a22.
|
||||||
|
- ✅ Все проверки и task-раннеры запускаются с `-I` (isolated): каталог запуска больше не перекрывает venv (был артефакт cwd-затенения исходниками). Затронуты: check, дефолтные раннеры агента, визард (config/save/start), serve, agent_cmd.
|
||||||
|
- ✅ Релиз v1.1.0-alpha.22; полный гейт (race + lint 0 issues + pytest 213); бинарник юзера обновлён, рантайм переустановлен (a22), воркер с concurrency=4; большой джоб юзера (1934 шарда) завершён (очередь 0).
|
||||||
|
|
||||||
## Completion
|
## Completion
|
||||||
COMPLETED — ночной план выполнен полностью (10 пунктов + 2 найденных бага, включая E2E quorum на релизном коде). Все гейты зелёные, релиз v1.1.0-alpha.16 опубликован.
|
COMPLETED — ночной план выполнен полностью (10 пунктов + 2 найденных бага, включая E2E quorum на релизном коде). Все гейты зелёные, релиз v1.1.0-alpha.16 опубликован.
|
||||||
|
|
||||||
|
## Progress (день 2: контроль версий + изоляция)
|
||||||
|
- ✅ Визард сверяет установленную версию scimesh с версией бинарника: preflight показывает «installed X, worker needs Y — press Install» и снова предлагает кнопку Install (ensureMatchingScimeshVersion, dev-сборки пропускают). Проверено на машине: a21 vs a22 → красный → Install → зелёный 1.1.0a22.
|
||||||
|
- ✅ Все проверки и task-раннеры запускаются с `-I` (isolated): каталог запуска больше не перекрывает venv (был артефакт cwd-затенения исходниками). Затронуты: check, дефолтные раннеры агента, визард (config/save/start), serve, agent_cmd.
|
||||||
|
- ✅ Релиз v1.1.0-alpha.22; полный гейт (race + lint 0 issues + pytest 213); бинарник юзера обновлён, рантайм переустановлен (a22), воркер с concurrency=4; большой джоб юзера (1934 шарда) завершён (очередь 0).
|
||||||
|
|
||||||
## Completion (предыдущая задача)
|
## Completion (предыдущая задача)
|
||||||
COMPLETED — пайплайн доведён до рабочего состояния и проверен на релизных артефактах v1.1.0-alpha.14.
|
COMPLETED — пайплайн доведён до рабочего состояния и проверен на релизных артефактах v1.1.0-alpha.14.
|
||||||
|
|||||||
@@ -61,7 +61,24 @@ powershell -ExecutionPolicy Bypass -c "irm https://raw.githubusercontent.com/emi
|
|||||||
```
|
```
|
||||||
|
|
||||||
Set `SCIMESH_AUTO_START=0` to install without starting anything. The old demo
|
Set `SCIMESH_AUTO_START=0` to install without starting anything. The old demo
|
||||||
control room was removed: `/ui` is the admin console. A standalone
|
control room was removed: `/ui` is the admin console.
|
||||||
|
|
||||||
|
**HTTPS (TLS):** serve can encrypt everything with a self-signed certificate —
|
||||||
|
`coordinator serve --tls-autogen` generates one into the data directory and
|
||||||
|
prints its fingerprint; workers trust it via `SCIMESH_CA_CERT=<path>` (or the
|
||||||
|
explicit opt-in `SCIMESH_INSECURE_SKIP_VERIFY=1`). Custom certificates go
|
||||||
|
through `--tls-cert`/`--tls-key` (or `SCIMESH_TLS_CERT`/`SCIMESH_TLS_KEY`).
|
||||||
|
Without TLS, traffic on the LAN is plaintext. New UI accounts can be closed
|
||||||
|
with `--disable-registration` (or `SCIMESH_DISABLE_REGISTRATION=1`).
|
||||||
|
|
||||||
|
To remove a component, run the matching uninstaller (data is kept unless you
|
||||||
|
pass `--purge`):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.sh | bash -s coordinator
|
||||||
|
curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.sh | bash -s worker --purge
|
||||||
|
# Windows: irm .../uninstall.ps1 | iex (set $env:SCIMESH_COMPONENT, -Purge deletes data)
|
||||||
|
``` A standalone
|
||||||
worker is installed the same way (`bash -s worker`, or
|
worker is installed the same way (`bash -s worker`, or
|
||||||
`SCIMESH_COMPONENT=worker` on Windows); its installer opens the local setup
|
`SCIMESH_COMPONENT=worker` on Windows); its installer opens the local setup
|
||||||
wizard (`worker-agent setup`) in the browser automatically.
|
wizard (`worker-agent setup`) in the browser automatically.
|
||||||
|
|||||||
@@ -45,7 +45,7 @@ func runAgent(args []string) error {
|
|||||||
return fmt.Errorf("--coordinator-url, --token, and --work-dir are required")
|
return fmt.Errorf("--coordinator-url, --token, and --work-dir are required")
|
||||||
}
|
}
|
||||||
if *taskRunner == "" {
|
if *taskRunner == "" {
|
||||||
*taskRunner = "python -m scimesh.worker.task"
|
*taskRunner = "python -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||||||
|
|||||||
@@ -223,8 +223,16 @@ func runWithConfig(cfg infra.Config) error {
|
|||||||
|
|
||||||
// deps.ready backs /health: readiness means the database answers, not just
|
// deps.ready backs /health: readiness means the database answers, not just
|
||||||
// that the process is alive.
|
// that the process is alive.
|
||||||
api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, cfg.JWTSecret, cfg.UserserviceURL, m, deps.ready, cfg.PublicCoordinatorURL, cfg.PublicUserserviceURL, cfg.DocsDir)
|
api := httptransport.NewServerWithOptions(
|
||||||
err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken))
|
useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes,
|
||||||
|
cfg.JWTSecret, cfg.UserserviceURL, m, deps.ready,
|
||||||
|
httptransport.ServerOptions{DisableRegistration: cfg.DisableRegistration},
|
||||||
|
cfg.PublicCoordinatorURL, cfg.PublicUserserviceURL, cfg.DocsDir)
|
||||||
|
var tlsOpts []infra.TLSConfig
|
||||||
|
if cfg.TLSCertFile != "" && cfg.TLSKeyFile != "" {
|
||||||
|
tlsOpts = []infra.TLSConfig{{CertFile: cfg.TLSCertFile, KeyFile: cfg.TLSKeyFile}}
|
||||||
|
}
|
||||||
|
err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken), tlsOpts...)
|
||||||
|
|
||||||
// Shutdown order matters, and defers alone cannot express it (they run
|
// Shutdown order matters, and defers alone cannot express it (they run
|
||||||
// LIFO, so the deferred stop() would fire *after* the wait below).
|
// LIFO, so the deferred stop() would fire *after* the wait below).
|
||||||
|
|||||||
@@ -6,10 +6,17 @@ import (
|
|||||||
"github.com/emil28092005/SciMesh/coordinator/internal/agent"
|
"github.com/emil28092005/SciMesh/coordinator/internal/agent"
|
||||||
|
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
|
"crypto/rsa"
|
||||||
|
"crypto/sha256"
|
||||||
|
"crypto/x509"
|
||||||
|
"crypto/x509/pkix"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
|
"encoding/pem"
|
||||||
"flag"
|
"flag"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"math/big"
|
||||||
|
"net"
|
||||||
"os"
|
"os"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -42,6 +49,10 @@ func runServe(args []string) error {
|
|||||||
email = flags.String("admin-email", "admin@scimesh.local", "admin account email")
|
email = flags.String("admin-email", "admin@scimesh.local", "admin account email")
|
||||||
password = flags.String("admin-password", "", "admin password (generated on first run when empty)")
|
password = flags.String("admin-password", "", "admin password (generated on first run when empty)")
|
||||||
publicURL = flags.String("public-url", "", "browser/worker-facing coordinator URL (default: http://<addr>)")
|
publicURL = flags.String("public-url", "", "browser/worker-facing coordinator URL (default: http://<addr>)")
|
||||||
|
tlsCert = flags.String("tls-cert", "", "TLS certificate file (enables HTTPS together with --tls-key)")
|
||||||
|
tlsKey = flags.String("tls-key", "", "TLS private key file")
|
||||||
|
tlsGen = flags.Bool("tls-autogen", false, "generate a self-signed certificate in the data dir and serve HTTPS")
|
||||||
|
noReg = flags.Bool("disable-registration", false, "forbid new UI accounts")
|
||||||
)
|
)
|
||||||
if err := flags.Parse(args); err != nil {
|
if err := flags.Parse(args); err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -96,27 +107,50 @@ func runServe(args []string) error {
|
|||||||
defer func() { _ = closeUsers() }()
|
defer func() { _ = closeUsers() }()
|
||||||
|
|
||||||
// 5. Local worker agents before the server, so they can claim immediately.
|
// 5. Local worker agents before the server, so they can claim immediately.
|
||||||
coordinatorURL := "http://" + *addr
|
// They always dial the loopback address: an --addr of 0.0.0.0 is not a
|
||||||
agents, err := spawnAgents(ctx, log, *dataDir, *workers, coordinatorURL, workerToken, venvPython)
|
// connectable target from the same host.
|
||||||
|
agentURL, resolvedPublic := serveURLs(*addr, *publicURL)
|
||||||
|
agents, err := spawnAgents(ctx, log, *dataDir, *workers, agentURL, workerToken, venvPython)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer stopAgents(agents)
|
defer stopAgents(agents)
|
||||||
|
|
||||||
// 6. The coordinator server itself.
|
// 6. The coordinator server itself.
|
||||||
coordinatorPublicURL := *publicURL
|
// TLS: explicit cert/key win; --tls-autogen creates a self-signed pair in
|
||||||
if coordinatorPublicURL == "" {
|
// the data dir on first use (fingerprint printed for pinning).
|
||||||
coordinatorPublicURL = "http://" + *addr
|
tlsCertFile, tlsKeyFile := *tlsCert, *tlsKey
|
||||||
|
if tlsCertFile == "" && tlsKeyFile == "" && *tlsGen {
|
||||||
|
tlsCertFile = filepath.Join(*dataDir, "tls.crt")
|
||||||
|
tlsKeyFile = filepath.Join(*dataDir, "tls.key")
|
||||||
|
if _, err := os.Stat(tlsCertFile); err != nil {
|
||||||
|
fingerprint, err := generateSelfSigned(tlsCertFile, tlsKeyFile, *dataDir, *addr)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("generate TLS certificate: %w", err)
|
||||||
|
}
|
||||||
|
log.Info("generated a self-signed TLS certificate", "cert", tlsCertFile, "fingerprint", fingerprint)
|
||||||
|
fmt.Printf("TLS: self-signed certificate generated (SHA-256 fingerprint %s).\n", fingerprint)
|
||||||
|
fmt.Printf("Trust it on workers with SCIMESH_CA_CERT=%s (or SCIMESH_INSECURE_SKIP_VERIFY=1).\n", tlsCertFile)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
if (tlsCertFile == "") != (tlsKeyFile == "") {
|
||||||
|
return fmt.Errorf("--tls-cert and --tls-key must be provided together")
|
||||||
|
}
|
||||||
|
|
||||||
cfg := infra.Config{
|
cfg := infra.Config{
|
||||||
Addr: *addr,
|
Addr: *addr,
|
||||||
DatabaseEngine: "sqlite",
|
DatabaseEngine: "sqlite",
|
||||||
DBPath: filepath.Join(*dataDir, "scimesh.db"),
|
DBPath: filepath.Join(*dataDir, "scimesh.db"),
|
||||||
|
TLSCertFile: tlsCertFile,
|
||||||
|
TLSKeyFile: tlsKeyFile,
|
||||||
|
DisableRegistration: *noReg || os.Getenv("SCIMESH_DISABLE_REGISTRATION") == "1",
|
||||||
Token: workerToken,
|
Token: workerToken,
|
||||||
JWTSecret: jwtSecret,
|
JWTSecret: jwtSecret,
|
||||||
UserserviceURL: "http://" + usersAddr,
|
UserserviceURL: "http://" + usersAddr,
|
||||||
PublicCoordinatorURL: coordinatorPublicURL,
|
PublicCoordinatorURL: resolvedPublic,
|
||||||
PublicUserserviceURL: "http://" + usersAddr,
|
// The exchange is fronted by the coordinator's own proxy, so the UI
|
||||||
|
// falls back to the coordinator origin for USERSERVICE_URL.
|
||||||
|
PublicUserserviceURL: "",
|
||||||
LogLevel: "info",
|
LogLevel: "info",
|
||||||
StorageDir: filepath.Join(*dataDir, "artifacts"),
|
StorageDir: filepath.Join(*dataDir, "artifacts"),
|
||||||
DocsDir: *docsDir,
|
DocsDir: *docsDir,
|
||||||
@@ -132,12 +166,13 @@ func runServe(args []string) error {
|
|||||||
WorkerOfflineAfter: 1 * time.Minute,
|
WorkerOfflineAfter: 1 * time.Minute,
|
||||||
AutoMigrate: true,
|
AutoMigrate: true,
|
||||||
}
|
}
|
||||||
|
browserURL, _ := serveURLs(*addr, *publicURL)
|
||||||
if *open {
|
if *open {
|
||||||
openBrowser("http://" + *addr + "/ui/admin")
|
openBrowser(browserURL + "/ui/admin")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Print the login once the server is about to start.
|
// Print the login once the server is about to start.
|
||||||
fmt.Printf("\nSciMesh is starting at http://%s/ui\n", *addr)
|
fmt.Printf("\nSciMesh is starting at %s/ui\n", browserURL)
|
||||||
fmt.Printf(" admin login: %s / %s\n", *email, *password)
|
fmt.Printf(" admin login: %s / %s\n", *email, *password)
|
||||||
if runtimeStatus(venvPython) {
|
if runtimeStatus(venvPython) {
|
||||||
fmt.Printf(" scientific runtime: ready (%s)\n", venvPython)
|
fmt.Printf(" scientific runtime: ready (%s)\n", venvPython)
|
||||||
@@ -236,9 +271,9 @@ func stopAgents(agents []*exec.Cmd) {
|
|||||||
// system `python`.
|
// system `python`.
|
||||||
func defaultTaskRunner(venvPython string) string {
|
func defaultTaskRunner(venvPython string) string {
|
||||||
if runtimeStatus(venvPython) {
|
if runtimeStatus(venvPython) {
|
||||||
return venvPython + " -m scimesh.worker.task"
|
return venvPython + " -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
return "python -m scimesh.worker.task"
|
return "python -I -m scimesh.worker.task"
|
||||||
}
|
}
|
||||||
|
|
||||||
// ensureRuntime creates the managed venv and installs scimesh into it, unless
|
// ensureRuntime creates the managed venv and installs scimesh into it, unless
|
||||||
@@ -337,3 +372,102 @@ func openBrowser(target string) {
|
|||||||
// #nosec G204 -- target is the local UI URL the operator asked to open.
|
// #nosec G204 -- target is the local UI URL the operator asked to open.
|
||||||
_ = exec.CommandContext(context.Background(), command, target).Start()
|
_ = exec.CommandContext(context.Background(), command, target).Start()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// serveURLs derives the two addresses of a serve instance from the listen
|
||||||
|
// address and the optional --public-url flag:
|
||||||
|
//
|
||||||
|
// - the agent URL is always the loopback form of the port, because spawned
|
||||||
|
// local workers share the host and 0.0.0.0 is not connectable from it;
|
||||||
|
// - the public URL is what browsers and remote workers are told. An explicit
|
||||||
|
// --public-url wins; a listen host that is a real address is used as-is;
|
||||||
|
// a wildcard host (0.0.0.0, ::, or empty) yields an empty public URL, so
|
||||||
|
// the UI falls back to the browser's own origin (the coordinator's LAN
|
||||||
|
// address as the browser sees it).
|
||||||
|
func serveURLs(addr, publicURL string) (agentURL, resolvedPublic string) {
|
||||||
|
host, port, err := net.SplitHostPort(addr)
|
||||||
|
if err != nil {
|
||||||
|
// No port in the listen address: assume the default and treat the
|
||||||
|
// whole string as a host (e.g. a bare wildcard).
|
||||||
|
host, port = addr, "8080"
|
||||||
|
}
|
||||||
|
agentURL = "http://127.0.0.1:" + port
|
||||||
|
if publicURL != "" {
|
||||||
|
return agentURL, publicURL
|
||||||
|
}
|
||||||
|
host = strings.Trim(host, "[]")
|
||||||
|
switch host {
|
||||||
|
case "", "0.0.0.0", "::":
|
||||||
|
return agentURL, ""
|
||||||
|
default:
|
||||||
|
return agentURL, "http://" + addr
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// generateSelfSigned writes a self-signed certificate for the listen host and
|
||||||
|
// the machine's LAN addresses, so HTTPS works without a CA on a trusted
|
||||||
|
// network. The returned value is the certificate's SHA-256 fingerprint.
|
||||||
|
func generateSelfSigned(certPath, keyPath, dataDir, addr string) (string, error) {
|
||||||
|
host, _, err := net.SplitHostPort(addr)
|
||||||
|
if err != nil {
|
||||||
|
host = addr
|
||||||
|
}
|
||||||
|
host = strings.Trim(host, "[]")
|
||||||
|
ips := []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}
|
||||||
|
if parsed := net.ParseIP(host); parsed != nil && !parsed.IsUnspecified() {
|
||||||
|
ips = append(ips, parsed)
|
||||||
|
} else if host == "" || parsed != nil {
|
||||||
|
// Wildcard listen addresses: add every local interface address.
|
||||||
|
if addrs, err := net.InterfaceAddrs(); err == nil {
|
||||||
|
for _, a := range addrs {
|
||||||
|
if ipnet, ok := a.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
|
||||||
|
ips = append(ips, ipnet.IP)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
names := []string{"localhost", host}
|
||||||
|
if host != "" && host != "localhost" {
|
||||||
|
names = append(names, host)
|
||||||
|
}
|
||||||
|
key, err := rsa.GenerateKey(rand.Reader, 2048)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
template := x509.Certificate{
|
||||||
|
SerialNumber: serial,
|
||||||
|
Subject: pkix.Name{CommonName: "SciMesh coordinator"},
|
||||||
|
NotBefore: time.Now().Add(-time.Hour),
|
||||||
|
NotAfter: time.Now().AddDate(1, 0, 0),
|
||||||
|
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
|
||||||
|
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
||||||
|
DNSNames: names,
|
||||||
|
IPAddresses: ips,
|
||||||
|
}
|
||||||
|
der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
if err := os.MkdirAll(dataDir, 0o750); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
|
||||||
|
keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)})
|
||||||
|
// The certificate is shared with workers via SCIMESH_CA_CERT, so it must
|
||||||
|
// stay readable; the key stays private.
|
||||||
|
if err := os.WriteFile(certPath, certPEM, 0o644); err != nil { //nolint:gosec // G306: cert is public by design
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
sum := sha256.Sum256(der)
|
||||||
|
var parts []string
|
||||||
|
for _, b := range sum[:] {
|
||||||
|
parts = append(parts, fmt.Sprintf("%02x", b))
|
||||||
|
}
|
||||||
|
return strings.Join(parts, ":"), nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
func TestServeURLs(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
addr, public, agent, resolved string
|
||||||
|
}{
|
||||||
|
{"127.0.0.1:8080", "", "http://127.0.0.1:8080", "http://127.0.0.1:8080"},
|
||||||
|
{"0.0.0.0:8080", "", "http://127.0.0.1:8080", ""},
|
||||||
|
{":8080", "", "http://127.0.0.1:8080", ""},
|
||||||
|
{"::", "", "http://127.0.0.1:8080", ""},
|
||||||
|
{"192.168.1.10:8080", "", "http://127.0.0.1:8080", "http://192.168.1.10:8080"},
|
||||||
|
{"0.0.0.0:8080", "http://cluster.example:8080", "http://127.0.0.1:8080", "http://cluster.example:8080"},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
agent, resolved := serveURLs(c.addr, c.public)
|
||||||
|
if agent != c.agent || resolved != c.resolved {
|
||||||
|
t.Errorf("serveURLs(%q, %q) = (%q, %q), want (%q, %q)", c.addr, c.public, agent, resolved, c.agent, c.resolved)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -75,7 +75,7 @@ func (p *WorkerKeyToken) exchangeLocked() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
request.Header.Set("Content-Type", "application/json")
|
request.Header.Set("Content-Type", "application/json")
|
||||||
client := &http.Client{Timeout: p.timeout}
|
client := &http.Client{Timeout: p.timeout, Transport: tlsTransport(nil)}
|
||||||
response, err := client.Do(request)
|
response, err := client.Do(request)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("worker key exchange request failed")
|
return fmt.Errorf("worker key exchange request failed")
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ func checkHTTP(ctx context.Context, url string, timeout time.Duration) (CheckIte
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return CheckItem{Name: "coordinator", OK: false, Detail: "invalid URL"}, ""
|
return CheckItem{Name: "coordinator", OK: false, Detail: "invalid URL"}, ""
|
||||||
}
|
}
|
||||||
resp, err := http.DefaultClient.Do(req)
|
resp, err := (&http.Client{Timeout: timeout, Transport: tlsTransport(nil)}).Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
detail := err.Error()
|
detail := err.Error()
|
||||||
if strings.Contains(detail, "connection refused") {
|
if strings.Contains(detail, "connection refused") {
|
||||||
@@ -88,8 +88,12 @@ func CheckEnvironment(ctx context.Context) CheckReport {
|
|||||||
// execute with.
|
// execute with.
|
||||||
func CheckEnvironmentWithPython(ctx context.Context, python string) CheckReport {
|
func CheckEnvironmentWithPython(ctx context.Context, python string) CheckReport {
|
||||||
report := CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: true, Detail: python}}
|
report := CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: true, Detail: python}}
|
||||||
|
// The version comes from importlib.metadata, so the wizard can compare the
|
||||||
|
// installed package with the binary version and offer an upgrade. -I keeps
|
||||||
|
// the working directory out of sys.path, so a scimesh checkout in the
|
||||||
|
// wizard's cwd can never shadow the venv installation.
|
||||||
//nolint:gosec // G204: python is a resolved interpreter path, the argument list is constant
|
//nolint:gosec // G204: python is a resolved interpreter path, the argument list is constant
|
||||||
cmd := exec.CommandContext(ctx, python, "-c", "import scimesh; print(scimesh.__version__ if hasattr(scimesh, '__version__') else 'installed')")
|
cmd := exec.CommandContext(ctx, python, "-I", "-c", "import importlib.metadata as m; print(m.version('scimesh'))")
|
||||||
out, err := cmd.Output()
|
out, err := cmd.Output()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// The worker executes workloads by spawning scimesh's task runner, so
|
// The worker executes workloads by spawning scimesh's task runner, so
|
||||||
@@ -144,7 +148,7 @@ func CheckAuth(ctx context.Context, url, token, workerKey, userserviceURL string
|
|||||||
item.Detail = "no credential configured — will be checked at registration"
|
item.Detail = "no credential configured — will be checked at registration"
|
||||||
return item
|
return item
|
||||||
}
|
}
|
||||||
client := &http.Client{Timeout: 30 * time.Second}
|
client := &http.Client{Timeout: 30 * time.Second, Transport: tlsTransport(nil)}
|
||||||
if workerKey != "" && userserviceURL != "" {
|
if workerKey != "" && userserviceURL != "" {
|
||||||
payload, _ := json.Marshal(map[string]string{"key": workerKey})
|
payload, _ := json.Marshal(map[string]string{"key": workerKey})
|
||||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(userserviceURL, "/")+"/worker-tokens/exchange", strings.NewReader(string(payload)))
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(userserviceURL, "/")+"/worker-tokens/exchange", strings.NewReader(string(payload)))
|
||||||
|
|||||||
@@ -48,16 +48,24 @@ type Client struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func NewClient(baseURL string, tokens TokenProvider, timeout time.Duration) *Client {
|
func NewClient(baseURL string, tokens TokenProvider, timeout time.Duration) *Client {
|
||||||
|
// Payload transfers get a more generous budget than control calls: a large
|
||||||
|
// shard over a slow link easily outlives the API timeout.
|
||||||
|
transferTimeout := timeout * 4
|
||||||
|
if transferTimeout < 2*time.Minute {
|
||||||
|
transferTimeout = 2 * time.Minute
|
||||||
|
}
|
||||||
return &Client{
|
return &Client{
|
||||||
baseURL: strings.TrimRight(baseURL, "/"),
|
baseURL: strings.TrimRight(baseURL, "/"),
|
||||||
tokens: tokens,
|
tokens: tokens,
|
||||||
timeout: timeout,
|
timeout: timeout,
|
||||||
apiClient: &http.Client{
|
apiClient: &http.Client{
|
||||||
Timeout: timeout,
|
Timeout: timeout,
|
||||||
|
Transport: tlsTransport(nil),
|
||||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||||
},
|
},
|
||||||
dlClient: &http.Client{
|
dlClient: &http.Client{
|
||||||
Timeout: timeout,
|
Timeout: transferTimeout,
|
||||||
|
Transport: tlsTransport(nil),
|
||||||
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
||||||
if len(via) >= 10 {
|
if len(via) >= 10 {
|
||||||
return fmt.Errorf("too many redirects")
|
return fmt.Errorf("too many redirects")
|
||||||
|
|||||||
@@ -1,8 +1,10 @@
|
|||||||
package agent
|
package agent
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"encoding/pem"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -223,3 +225,72 @@ func sha256Of(t *testing.T, value string) string {
|
|||||||
digest := sha256.Sum256([]byte(value))
|
digest := sha256.Sum256([]byte(value))
|
||||||
return fmt.Sprintf("%x", digest)
|
return fmt.Sprintf("%x", digest)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestNewClientTransferTimeoutExceedsAPITimeout(t *testing.T) {
|
||||||
|
c := NewClient("http://coord:8080", &StaticToken{token: "t"}, 30*time.Second)
|
||||||
|
if c.apiClient.Timeout != 30*time.Second {
|
||||||
|
t.Errorf("api timeout = %v, want 30s", c.apiClient.Timeout)
|
||||||
|
}
|
||||||
|
if c.dlClient.Timeout < 2*time.Minute {
|
||||||
|
t.Errorf("transfer timeout = %v, want at least 2m", c.dlClient.Timeout)
|
||||||
|
}
|
||||||
|
short := NewClient("http://coord:8080", &StaticToken{token: "t"}, 3*time.Minute)
|
||||||
|
if short.dlClient.Timeout != 12*time.Minute {
|
||||||
|
t.Errorf("transfer timeout = %v, want 4x the api timeout", short.dlClient.Timeout)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTLSClientHonoursSkipVerify(t *testing.T) {
|
||||||
|
t.Setenv("SCIMESH_INSECURE_SKIP_VERIFY", "1")
|
||||||
|
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
_, _ = w.Write([]byte(`{"status":"ok"}`))
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
client := tlsClient(5 * time.Second)
|
||||||
|
req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL+"/health", nil)
|
||||||
|
resp, err := client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("TLS server must be reachable with skip-verify: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = resp.Body.Close() }()
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Errorf("status = %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTLSClientFailsWithoutTrust(t *testing.T) {
|
||||||
|
t.Setenv("SCIMESH_INSECURE_SKIP_VERIFY", "")
|
||||||
|
t.Setenv("SCIMESH_CA_CERT", "")
|
||||||
|
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||||
|
defer server.Close()
|
||||||
|
client := tlsClient(5 * time.Second)
|
||||||
|
req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL+"/health", nil)
|
||||||
|
if resp, err := client.Do(req); err == nil {
|
||||||
|
_ = resp.Body.Close()
|
||||||
|
t.Error("untrusted TLS server must fail verification")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTLSClientTrustsCAPool(t *testing.T) {
|
||||||
|
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
_, _ = w.Write([]byte("ok"))
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
ca := server.Certificate()
|
||||||
|
path := filepath.Join(t.TempDir(), "ca.pem")
|
||||||
|
if err := os.WriteFile(path, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.Raw}), 0o600); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Setenv("SCIMESH_CA_CERT", path)
|
||||||
|
t.Setenv("SCIMESH_INSECURE_SKIP_VERIFY", "")
|
||||||
|
client := tlsClient(5 * time.Second)
|
||||||
|
req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL+"/health", nil)
|
||||||
|
resp, err := client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CA-trusted TLS server must verify: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = resp.Body.Close() }()
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Errorf("status = %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -32,6 +32,10 @@ type Config struct {
|
|||||||
TaskRunner []string // command + args; defaults to python -m scimesh.worker.task
|
TaskRunner []string // command + args; defaults to python -m scimesh.worker.task
|
||||||
MaxTasks int // 0 = unlimited
|
MaxTasks int // 0 = unlimited
|
||||||
ExitWhenIdle bool
|
ExitWhenIdle bool
|
||||||
|
// Concurrency is how many claim→execute→upload loops run in parallel
|
||||||
|
// under one worker id: N shards processed concurrently on one machine,
|
||||||
|
// using the coordinator's own task pipeline as the parallel unit.
|
||||||
|
Concurrency int
|
||||||
}
|
}
|
||||||
|
|
||||||
func envList(name string) ([]string, error) {
|
func envList(name string) ([]string, error) {
|
||||||
@@ -108,7 +112,7 @@ func LoadConfig() (*Config, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(runner) == 0 {
|
if len(runner) == 0 {
|
||||||
runner = []string{"python", "-m", "scimesh.worker.task"}
|
runner = []string{"python", "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
maxTasks := 0
|
maxTasks := 0
|
||||||
if raw := os.Getenv("MAX_TASKS"); raw != "" {
|
if raw := os.Getenv("MAX_TASKS"); raw != "" {
|
||||||
@@ -145,9 +149,22 @@ func LoadConfig() (*Config, error) {
|
|||||||
TaskRunner: runner,
|
TaskRunner: runner,
|
||||||
MaxTasks: maxTasks,
|
MaxTasks: maxTasks,
|
||||||
ExitWhenIdle: os.Getenv("EXIT_WHEN_IDLE") == "1",
|
ExitWhenIdle: os.Getenv("EXIT_WHEN_IDLE") == "1",
|
||||||
|
Concurrency: envInt("WORKER_CONCURRENCY", 1),
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func envInt(name string, fallback int) int {
|
||||||
|
raw := os.Getenv(name)
|
||||||
|
if raw == "" {
|
||||||
|
return fallback
|
||||||
|
}
|
||||||
|
parsed, err := strconv.Atoi(raw)
|
||||||
|
if err != nil || parsed < 1 {
|
||||||
|
return fallback
|
||||||
|
}
|
||||||
|
return parsed
|
||||||
|
}
|
||||||
|
|
||||||
func durationEnv(name string, fallback time.Duration) (time.Duration, error) {
|
func durationEnv(name string, fallback time.Duration) (time.Duration, error) {
|
||||||
raw := os.Getenv(name)
|
raw := os.Getenv(name)
|
||||||
if raw == "" {
|
if raw == "" {
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ type ConfigFile struct {
|
|||||||
WorkerName string `json:"worker_name,omitempty"`
|
WorkerName string `json:"worker_name,omitempty"`
|
||||||
CPUCount int `json:"cpu_count"`
|
CPUCount int `json:"cpu_count"`
|
||||||
MemoryMB int `json:"memory_mb"`
|
MemoryMB int `json:"memory_mb"`
|
||||||
|
Concurrency int `json:"concurrency,omitempty"`
|
||||||
TaskRunner []string `json:"task_runner,omitempty"`
|
TaskRunner []string `json:"task_runner,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -103,11 +104,15 @@ func (f *ConfigFile) Config() (*Config, error) {
|
|||||||
if config.MemoryMB < 0 {
|
if config.MemoryMB < 0 {
|
||||||
config.MemoryMB = 0
|
config.MemoryMB = 0
|
||||||
}
|
}
|
||||||
|
config.Concurrency = f.Concurrency
|
||||||
|
if config.Concurrency < 1 {
|
||||||
|
config.Concurrency = 1
|
||||||
|
}
|
||||||
if len(f.TaskRunner) > 0 {
|
if len(f.TaskRunner) > 0 {
|
||||||
config.TaskRunner = f.TaskRunner
|
config.TaskRunner = f.TaskRunner
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) == 0 {
|
if len(config.TaskRunner) == 0 {
|
||||||
config.TaskRunner = []string{"python", "-m", "scimesh.worker.task"}
|
config.TaskRunner = []string{"python", "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
config.PollInterval = 2 * time.Second
|
config.PollInterval = 2 * time.Second
|
||||||
config.RequestTimeout = 30 * time.Second
|
config.RequestTimeout = 30 * time.Second
|
||||||
|
|||||||
@@ -37,15 +37,52 @@ func NewDaemon(config *Config, client *Client, runner *TaskRunner, log *slog.Log
|
|||||||
return &Daemon{config: config, client: client, runner: runner, log: log}
|
return &Daemon{config: config, client: client, runner: runner, log: log}
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunForever loops until interrupted, idle-exit, or max tasks.
|
// RunForever registers once, then runs the claim→execute→upload loop
|
||||||
|
// concurrently under one worker id. With Concurrency > 1, several shards are
|
||||||
|
// processed in parallel on this machine, using the coordinator's own task
|
||||||
|
// pipeline as the parallel unit.
|
||||||
func (d *Daemon) RunForever() error {
|
func (d *Daemon) RunForever() error {
|
||||||
|
if !d.registered {
|
||||||
|
if err := d.register(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
workers := d.config.Concurrency
|
||||||
|
if workers < 1 {
|
||||||
|
workers = 1
|
||||||
|
}
|
||||||
|
if workers == 1 {
|
||||||
|
return d.loop()
|
||||||
|
}
|
||||||
|
d.log.Info("agent running concurrently", "loops", workers)
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
errs := make(chan error, workers)
|
||||||
|
for i := 0; i < workers; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func(loop int) {
|
||||||
|
defer wg.Done()
|
||||||
|
if err := d.loop(); err != nil {
|
||||||
|
errs <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
errs <- nil
|
||||||
|
}(i)
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
close(errs)
|
||||||
|
for err := range errs {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// loop is one claim→execute→upload cycle until interrupted, idle-exit, or
|
||||||
|
// the shared max-tasks budget is consumed.
|
||||||
|
func (d *Daemon) loop() error {
|
||||||
failures := 0
|
failures := 0
|
||||||
for {
|
for {
|
||||||
if !d.registered {
|
|
||||||
if err := d.register(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
d.cleanupExpiredDirectories()
|
d.cleanupExpiredDirectories()
|
||||||
outcome, err := d.runOnce()
|
outcome, err := d.runOnce()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -63,8 +100,11 @@ func (d *Daemon) RunForever() error {
|
|||||||
}
|
}
|
||||||
failures = 0
|
failures = 0
|
||||||
if outcome.Claimed && outcome.Completed {
|
if outcome.Claimed && outcome.Completed {
|
||||||
|
d.mu.Lock()
|
||||||
d.completed++
|
d.completed++
|
||||||
if d.config.MaxTasks > 0 && d.completed >= d.config.MaxTasks {
|
done := d.config.MaxTasks > 0 && d.completed >= d.config.MaxTasks
|
||||||
|
d.mu.Unlock()
|
||||||
|
if done {
|
||||||
d.log.Info("max tasks reached")
|
d.log.Info("max tasks reached")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@@ -55,6 +56,7 @@ type fakeCoordinator struct {
|
|||||||
uploadSize int64
|
uploadSize int64
|
||||||
inputBytes []byte
|
inputBytes []byte
|
||||||
conflict bool // 409 on heartbeat/upload/result
|
conflict bool // 409 on heartbeat/upload/result
|
||||||
|
mu sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
func newFakeCoordinator(t *testing.T, task map[string]any) *fakeCoordinator {
|
func newFakeCoordinator(t *testing.T, task map[string]any) *fakeCoordinator {
|
||||||
@@ -64,6 +66,8 @@ func newFakeCoordinator(t *testing.T, task map[string]any) *fakeCoordinator {
|
|||||||
fake.uploadSize = int64(len(fake.inputBytes))
|
fake.uploadSize = int64(len(fake.inputBytes))
|
||||||
var server *httptest.Server
|
var server *httptest.Server
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
fake.mu.Lock()
|
||||||
|
defer fake.mu.Unlock()
|
||||||
switch {
|
switch {
|
||||||
case r.Method == http.MethodPost && r.URL.Path == "/workers/register":
|
case r.Method == http.MethodPost && r.URL.Path == "/workers/register":
|
||||||
writeJSON(w, http.StatusCreated, map[string]any{
|
writeJSON(w, http.StatusCreated, map[string]any{
|
||||||
@@ -294,3 +298,78 @@ func TestDaemonIdleClaimIsNotCompleted(t *testing.T) {
|
|||||||
t.Fatalf("outcome = %+v", outcome)
|
t.Fatalf("outcome = %+v", outcome)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDaemonConcurrencyProcessesTasksInParallel(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
marker := filepath.Join(t.TempDir(), "marker")
|
||||||
|
script := filepath.Join(t.TempDir(), "fake-runner.sh")
|
||||||
|
content := `#!/bin/sh
|
||||||
|
out=""
|
||||||
|
task_dir=""
|
||||||
|
while [ "$#" -gt 0 ]; do
|
||||||
|
case "$1" in
|
||||||
|
--output) out="$2"; shift 2;;
|
||||||
|
--task-dir) task_dir="$2"; shift 2;;
|
||||||
|
*) shift;;
|
||||||
|
esac
|
||||||
|
done
|
||||||
|
echo start >> ` + marker + `
|
||||||
|
sleep 1
|
||||||
|
echo end >> ` + marker + `
|
||||||
|
printf 'id,score\n1,1\n' > "$task_dir/result.csv"
|
||||||
|
printf '{"artifact_path":"%s/result.csv","content_type":"text/csv","metrics":{"rows":1}}' "$task_dir" > "$out"
|
||||||
|
exit 0
|
||||||
|
`
|
||||||
|
if err := os.WriteFile(script, []byte(content), 0o755); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
fake := newFakeCoordinator(t, validClaimedTaskPayload())
|
||||||
|
defer fake.close()
|
||||||
|
config := &Config{
|
||||||
|
CoordinatorURL: fake.server.URL,
|
||||||
|
WorkerName: "concurrent-worker",
|
||||||
|
WorkerID: "22222222-2222-4222-8222-222222222222",
|
||||||
|
WorkDir: t.TempDir(),
|
||||||
|
CPUCount: 1,
|
||||||
|
PollInterval: time.Millisecond,
|
||||||
|
RequestTimeout: 5 * time.Second,
|
||||||
|
Heartbeat: 15 * time.Second,
|
||||||
|
Capabilities: []string{"similarity-search"},
|
||||||
|
TaskRunner: []string{script},
|
||||||
|
MaxTasks: 3,
|
||||||
|
Concurrency: 3,
|
||||||
|
}
|
||||||
|
client := NewClient(fake.server.URL, &StaticToken{token: "test-token"}, 5*time.Second)
|
||||||
|
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||||
|
daemon := NewDaemon(config, client, NewTaskRunner(config.TaskRunner), logger)
|
||||||
|
if err := daemon.RunForever(); err != nil {
|
||||||
|
t.Fatalf("run: %v", err)
|
||||||
|
}
|
||||||
|
raw, err := os.ReadFile(marker)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marker: %v", err)
|
||||||
|
}
|
||||||
|
starts := strings.Count(string(raw), "start\n")
|
||||||
|
ends := strings.Count(string(raw), "end\n")
|
||||||
|
if starts < 3 || ends < 3 {
|
||||||
|
t.Fatalf("marker: %d starts / %d ends, want at least 3/3", starts, ends)
|
||||||
|
}
|
||||||
|
// With three loops sleeping 1s each, the marker proves all three ran
|
||||||
|
// concurrently (three starts before the first end completes a 1s sleep).
|
||||||
|
lines := strings.Split(strings.TrimSpace(string(raw)), "\n")
|
||||||
|
concurrent := 0
|
||||||
|
running := 0
|
||||||
|
for _, line := range lines {
|
||||||
|
if line == "start" {
|
||||||
|
running++
|
||||||
|
if running > concurrent {
|
||||||
|
concurrent = running
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
running--
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if concurrent < 3 {
|
||||||
|
t.Errorf("max concurrent executions = %d, want 3", concurrent)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -323,7 +323,7 @@ func (s *Server) ensureVenvTaskRunner() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if venv := s.venvPython(); venv != "" {
|
if venv := s.venvPython(); venv != "" {
|
||||||
file.TaskRunner = []string{venv, "-m", "scimesh.worker.task"}
|
file.TaskRunner = []string{venv, "-I", "-m", "scimesh.worker.task"}
|
||||||
if payload, err := json.MarshalIndent(file, "", " "); err == nil {
|
if payload, err := json.MarshalIndent(file, "", " "); err == nil {
|
||||||
_ = os.WriteFile(s.cfgPath, append(payload, '\n'), 0o600)
|
_ = os.WriteFile(s.cfgPath, append(payload, '\n'), 0o600)
|
||||||
}
|
}
|
||||||
@@ -377,6 +377,7 @@ type saveConfigRequest struct {
|
|||||||
WorkerName string `json:"worker_name"`
|
WorkerName string `json:"worker_name"`
|
||||||
CPUCount int `json:"cpu_count"`
|
CPUCount int `json:"cpu_count"`
|
||||||
MemoryMB int `json:"memory_mb"`
|
MemoryMB int `json:"memory_mb"`
|
||||||
|
Concurrency int `json:"concurrency"`
|
||||||
TaskRunner []string `json:"task_runner"`
|
TaskRunner []string `json:"task_runner"`
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -395,6 +396,7 @@ func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) {
|
|||||||
WorkerName: strings.TrimSpace(req.WorkerName),
|
WorkerName: strings.TrimSpace(req.WorkerName),
|
||||||
CPUCount: req.CPUCount,
|
CPUCount: req.CPUCount,
|
||||||
MemoryMB: req.MemoryMB,
|
MemoryMB: req.MemoryMB,
|
||||||
|
Concurrency: req.Concurrency,
|
||||||
TaskRunner: req.TaskRunner,
|
TaskRunner: req.TaskRunner,
|
||||||
}
|
}
|
||||||
if file.CoordinatorURL == "" {
|
if file.CoordinatorURL == "" {
|
||||||
@@ -418,12 +420,15 @@ func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) {
|
|||||||
if file.CPUCount < 1 {
|
if file.CPUCount < 1 {
|
||||||
file.CPUCount = 1
|
file.CPUCount = 1
|
||||||
}
|
}
|
||||||
|
if file.Concurrency < 1 {
|
||||||
|
file.Concurrency = 1
|
||||||
|
}
|
||||||
// The wizard UI bakes the venv python into the runner after an install;
|
// The wizard UI bakes the venv python into the runner after an install;
|
||||||
// an API-driven or scripted flow may not, so the server guarantees it:
|
// an API-driven or scripted flow may not, so the server guarantees it:
|
||||||
// workloads execute through scimesh's task runner, which lives in the venv.
|
// workloads execute through scimesh's task runner, which lives in the venv.
|
||||||
if len(file.TaskRunner) == 0 {
|
if len(file.TaskRunner) == 0 {
|
||||||
if venv := s.venvPython(); venv != "" {
|
if venv := s.venvPython(); venv != "" {
|
||||||
file.TaskRunner = []string{venv, "-m", "scimesh.worker.task"}
|
file.TaskRunner = []string{venv, "-I", "-m", "scimesh.worker.task"}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := agent.SaveConfigFile(s.cfgPath, file); err != nil {
|
if err := agent.SaveConfigFile(s.cfgPath, file); err != nil {
|
||||||
@@ -449,6 +454,9 @@ func (s *Server) handleTest(w http.ResponseWriter, r *http.Request) {
|
|||||||
// checking the bare system python3 would keep reporting scimesh as
|
// checking the bare system python3 would keep reporting scimesh as
|
||||||
// missing even though the worker would run with the venv.
|
// missing even though the worker would run with the venv.
|
||||||
report := agent.RunCheck(r.Context(), url, s.venvPython(), req.Token, req.WorkerKey, req.UserserviceURL)
|
report := agent.RunCheck(r.Context(), url, s.venvPython(), req.Token, req.WorkerKey, req.UserserviceURL)
|
||||||
|
if report.Scimesh.OK {
|
||||||
|
report.Scimesh = ensureMatchingScimeshVersion(report.Scimesh)
|
||||||
|
}
|
||||||
writeJSON(w, http.StatusOK, report)
|
writeJSON(w, http.StatusOK, report)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -623,3 +631,26 @@ func truncate(s string, n int) string {
|
|||||||
}
|
}
|
||||||
return s[:n] + "…"
|
return s[:n] + "…"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ensureMatchingScimeshVersion flips a green scimesh check to a stale one when
|
||||||
|
// the installed package does not match the worker-agent's own version: a
|
||||||
|
// version-locked wheel is the only supported runtime, and a mismatch means the
|
||||||
|
// workload catalog the worker advertises is not what it executes. The wizard
|
||||||
|
// UI then offers the Install button again. Dev builds have no release wheel,
|
||||||
|
// so they skip the comparison.
|
||||||
|
func ensureMatchingScimeshVersion(item agent.CheckItem) agent.CheckItem {
|
||||||
|
if agent.Version == "" || agent.Version == "dev" {
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
want := agent.NormalizePEP440(agent.Version)
|
||||||
|
got := strings.TrimSpace(item.Detail)
|
||||||
|
if got == "" || got == want {
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
item.OK = false
|
||||||
|
item.Detail = fmt.Sprintf(
|
||||||
|
"installed scimesh %s, but this worker-agent (%s) needs %s — press Install to upgrade",
|
||||||
|
got, agent.Version, want,
|
||||||
|
)
|
||||||
|
return item
|
||||||
|
}
|
||||||
|
|||||||
@@ -429,8 +429,8 @@ func TestStartPinsTheVenvTaskRunner(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) != 3 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-m" || config.TaskRunner[2] != "scimesh.worker.task" {
|
if len(config.TaskRunner) != 4 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-I" || config.TaskRunner[2] != "-m" || config.TaskRunner[3] != "scimesh.worker.task" {
|
||||||
t.Errorf("task runner = %v, want the venv python runner", config.TaskRunner)
|
t.Errorf("task runner = %v, want the venv python runner with -I", config.TaskRunner)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -450,8 +450,8 @@ func TestSaveConfigPinsVenvRunnerWhenPresent(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(config.TaskRunner) != 3 || config.TaskRunner[0] != venvPython {
|
if len(config.TaskRunner) != 4 || config.TaskRunner[0] != venvPython || config.TaskRunner[1] != "-I" {
|
||||||
t.Errorf("task runner = %v, want the venv python", config.TaskRunner)
|
t.Errorf("task runner = %v, want the venv python with -I", config.TaskRunner)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -462,7 +462,7 @@ func TestTestProbesTheVenvPythonAfterInstall(t *testing.T) {
|
|||||||
// a fake scimesh version so the preflight goes green through the venv.
|
// a fake scimesh version so the preflight goes green through the venv.
|
||||||
venvPython := filepath.Join(server.dir, "venv", "bin", "python")
|
venvPython := filepath.Join(server.dir, "venv", "bin", "python")
|
||||||
_ = os.MkdirAll(filepath.Dir(venvPython), 0o755)
|
_ = os.MkdirAll(filepath.Dir(venvPython), 0o755)
|
||||||
_ = os.WriteFile(venvPython, []byte("#!/bin/sh\nif [ \"$1\" = \"-c\" ]; then echo 9.9.9-test; exit 0; fi\nexit 0\n"), 0o755)
|
_ = os.WriteFile(venvPython, []byte("#!/bin/sh\nfor a in \"$@\"; do if [ \"$a\" = \"-c\" ]; then echo 9.9.9-test; exit 0; fi; done\nexit 0\n"), 0o755)
|
||||||
|
|
||||||
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, base+"/api/test", strings.NewReader(`{"coordinator_url":"http://127.0.0.1:1"}`))
|
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, base+"/api/test", strings.NewReader(`{"coordinator_url":"http://127.0.0.1:1"}`))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
@@ -497,3 +497,26 @@ time=6 level=WARN msg="agent cycle failed" error="boom"
|
|||||||
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
|
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func testCheckScimeshVersion(t *testing.T, installed, binary string, wantOK bool, wantDetail string) {
|
||||||
|
t.Helper()
|
||||||
|
old := agent.Version
|
||||||
|
agent.Version = binary
|
||||||
|
t.Cleanup(func() { agent.Version = old })
|
||||||
|
item := ensureMatchingScimeshVersion(agent.CheckItem{Name: "scimesh", OK: true, Detail: installed})
|
||||||
|
if item.OK != wantOK {
|
||||||
|
t.Errorf("installed=%s binary=%s: ok=%v, want %v (%s)", installed, binary, item.OK, wantOK, item.Detail)
|
||||||
|
}
|
||||||
|
if wantDetail != "" && !strings.Contains(item.Detail, wantDetail) {
|
||||||
|
t.Errorf("detail = %q, want it to contain %q", item.Detail, wantDetail)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsureMatchingScimeshVersion(t *testing.T) {
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a20", "1.1.0-alpha.20", true, "")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a17", "1.1.0-alpha.20", false, "press Install to upgrade")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a16.dev7+gea0fb8c59.d20260803", "1.1.0-alpha.20", false, "needs 1.1.0a20")
|
||||||
|
// Dev builds and unknown versions never block.
|
||||||
|
testCheckScimeshVersion(t, "anything", "dev", true, "")
|
||||||
|
testCheckScimeshVersion(t, "1.1.0a20", "", true, "")
|
||||||
|
}
|
||||||
|
|||||||
@@ -141,6 +141,7 @@ code{font-family:var(--mono);font-size:.86em}
|
|||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
<div class="field" id="cpu-field" style="display:none"><label>CPU count</label><input id="in-cpu" type="number" min="1" value="1"></div>
|
<div class="field" id="cpu-field" style="display:none"><label>CPU count</label><input id="in-cpu" type="number" min="1" value="1"></div>
|
||||||
|
<div class="field"><label>Concurrent task loops</label><input id="in-conc" type="number" min="1" max="64" value="1"><p class="hint">Process this many shards in parallel on this machine. Each loop runs its own task runner subprocess.</p></div>
|
||||||
<div class="actions"><button class="btn btn-ghost" id="b2b">← Back</button><button class="btn btn-primary" id="b2">Continue →</button></div>
|
<div class="actions"><button class="btn btn-ghost" id="b2b">← Back</button><button class="btn btn-primary" id="b2">Continue →</button></div>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
@@ -228,9 +229,10 @@ function draftConfig(){
|
|||||||
userservice_url:state.mode==='key'?$('in-users').value.trim():'',
|
userservice_url:state.mode==='key'?$('in-users').value.trim():'',
|
||||||
work_dir:$('in-dir').value.trim(),
|
work_dir:$('in-dir').value.trim(),
|
||||||
worker_name:$('in-name').value.trim(),
|
worker_name:$('in-name').value.trim(),
|
||||||
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0
|
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0,
|
||||||
|
concurrency:parseInt($('in-conc').value||'1',10)
|
||||||
};
|
};
|
||||||
if(state.venvPython)cfg.task_runner=[state.venvPython,'-m','scimesh.worker.task'];
|
if(state.venvPython)cfg.task_runner=[state.venvPython,'-I','-m','scimesh.worker.task'];
|
||||||
return cfg;
|
return cfg;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,69 @@
|
|||||||
|
package agent
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/tls"
|
||||||
|
"crypto/x509"
|
||||||
|
"log/slog"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"os"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// tlsClient builds an HTTP client whose transport trusts the coordinator's
|
||||||
|
// TLS certificate:
|
||||||
|
//
|
||||||
|
// - SCIMESH_CA_CERT=/path/to/ca.pem adds a root CA (for self-signed or
|
||||||
|
// private-CA coordinators);
|
||||||
|
// - SCIMESH_INSECURE_SKIP_VERIFY=1 disables verification entirely — only
|
||||||
|
// for trusted LANs where a self-signed certificate was auto-generated.
|
||||||
|
//
|
||||||
|
// Both settings are deliberately opt-in and noisy: a coordinator without them
|
||||||
|
// fails to verify, never silently downgrades.
|
||||||
|
func tlsClient(timeout time.Duration) *http.Client {
|
||||||
|
return &http.Client{Timeout: timeout, Transport: tlsTransport(nil)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// tlsTransport configures a transport honouring the trust environment.
|
||||||
|
func tlsTransport(base *http.Transport) *http.Transport {
|
||||||
|
if base == nil {
|
||||||
|
base = &http.Transport{
|
||||||
|
Proxy: http.ProxyFromEnvironment,
|
||||||
|
DialContext: (&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
|
||||||
|
MaxIdleConns: 100,
|
||||||
|
IdleConnTimeout: 90 * time.Second,
|
||||||
|
TLSHandshakeTimeout: 10 * time.Second,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
caPath := os.Getenv("SCIMESH_CA_CERT")
|
||||||
|
skip := os.Getenv("SCIMESH_INSECURE_SKIP_VERIFY") == "1"
|
||||||
|
if caPath == "" && !skip {
|
||||||
|
return base
|
||||||
|
}
|
||||||
|
tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12} //nolint:gosec // G402: min TLS 1.2 by default
|
||||||
|
if caPath != "" {
|
||||||
|
//nolint:gosec // G304: SCIMESH_CA_CERT is operator-configured
|
||||||
|
pem, err := os.ReadFile(caPath)
|
||||||
|
if err != nil {
|
||||||
|
slog.Warn("could not read SCIMESH_CA_CERT", "path", caPath, "err", err)
|
||||||
|
return base
|
||||||
|
}
|
||||||
|
pool, err := x509.SystemCertPool()
|
||||||
|
if err != nil {
|
||||||
|
pool = x509.NewCertPool()
|
||||||
|
}
|
||||||
|
if !pool.AppendCertsFromPEM(pem) {
|
||||||
|
slog.Warn("SCIMESH_CA_CERT contained no usable certificates", "path", caPath)
|
||||||
|
return base
|
||||||
|
}
|
||||||
|
tlsConfig.RootCAs = pool
|
||||||
|
}
|
||||||
|
if skip {
|
||||||
|
// G402 is about production code paths; here the operator explicitly
|
||||||
|
// opts into an unverified LAN trust root, so the bypass is intended.
|
||||||
|
tlsConfig.InsecureSkipVerify = true //nolint:gosec // G402: operator opt-in for self-signed LAN certs
|
||||||
|
slog.Warn("SCIMESH_INSECURE_SKIP_VERIFY=1: TLS certificate verification is disabled")
|
||||||
|
}
|
||||||
|
base.TLSClientConfig = tlsConfig
|
||||||
|
return base
|
||||||
|
}
|
||||||
@@ -55,6 +55,14 @@ type Config struct {
|
|||||||
// Directory of the built MkDocs site (site/) served at /ui/docs/. Empty
|
// Directory of the built MkDocs site (site/) served at /ui/docs/. Empty
|
||||||
// disables the docs route; the UI shows a hint page instead.
|
// disables the docs route; the UI shows a hint page instead.
|
||||||
DocsDir string
|
DocsDir string
|
||||||
|
// TLSCertFile and TLSKeyFile enable HTTPS when both are set. Self-signed
|
||||||
|
// certificates are fine for a trusted LAN; workers then need
|
||||||
|
// SCIMESH_CA_CERT or SCIMESH_INSECURE_SKIP_VERIFY to connect.
|
||||||
|
TLSCertFile string
|
||||||
|
TLSKeyFile string
|
||||||
|
// DisableRegistration forbids new UI accounts; the bootstrap admin still
|
||||||
|
// works. Existing accounts and worker keys are unaffected.
|
||||||
|
DisableRegistration bool
|
||||||
// Upper bound on an uploaded dataset or artifact body, in bytes.
|
// Upper bound on an uploaded dataset or artifact body, in bytes.
|
||||||
MaxUploadBytes int64
|
MaxUploadBytes int64
|
||||||
|
|
||||||
@@ -123,6 +131,9 @@ func LoadConfig() (Config, error) {
|
|||||||
LogFile: os.Getenv("LOG_FILE"),
|
LogFile: os.Getenv("LOG_FILE"),
|
||||||
StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"),
|
StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"),
|
||||||
DocsDir: os.Getenv("SCIMESH_DOCS_DIR"),
|
DocsDir: os.Getenv("SCIMESH_DOCS_DIR"),
|
||||||
|
TLSCertFile: os.Getenv("SCIMESH_TLS_CERT"),
|
||||||
|
TLSKeyFile: os.Getenv("SCIMESH_TLS_KEY"),
|
||||||
|
DisableRegistration: os.Getenv("SCIMESH_DISABLE_REGISTRATION") == "1",
|
||||||
MaxUploadBytes: 1 << 30, // 1 GiB
|
MaxUploadBytes: 1 << 30, // 1 GiB
|
||||||
DBMaxConns: 10,
|
DBMaxConns: 10,
|
||||||
DBConnectTimeout: 30 * time.Second,
|
DBConnectTimeout: 30 * time.Second,
|
||||||
|
|||||||
@@ -13,17 +13,25 @@ import (
|
|||||||
const shutdownGrace = 15 * time.Second
|
const shutdownGrace = 15 * time.Second
|
||||||
|
|
||||||
// Run serves handler until ctx is cancelled, then drains in-flight requests.
|
// Run serves handler until ctx is cancelled, then drains in-flight requests.
|
||||||
func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler) error {
|
func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler, tls ...TLSConfig) error {
|
||||||
srv := &http.Server{
|
srv := &http.Server{
|
||||||
Addr: addr,
|
Addr: addr,
|
||||||
Handler: handler,
|
Handler: handler,
|
||||||
ReadHeaderTimeout: 5 * time.Second,
|
ReadHeaderTimeout: 5 * time.Second,
|
||||||
}
|
}
|
||||||
|
secure := len(tls) > 0 && tls[0].CertFile != "" && tls[0].KeyFile != ""
|
||||||
|
|
||||||
// Buffered so this goroutine can exit even when nobody reads the channel
|
// Buffered so this goroutine can exit even when nobody reads the channel
|
||||||
// (the ctx.Done branch below) — an unbuffered send would leak it forever.
|
// (the ctx.Done branch below) — an unbuffered send would leak it forever.
|
||||||
errCh := make(chan error, 1)
|
errCh := make(chan error, 1)
|
||||||
go func() {
|
go func() {
|
||||||
|
if secure {
|
||||||
|
log.Info("coordinator listening (https)", "addr", addr)
|
||||||
|
if err := srv.ListenAndServeTLS(tls[0].CertFile, tls[0].KeyFile); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||||
|
errCh <- err
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
log.Info("coordinator listening", "addr", addr)
|
log.Info("coordinator listening", "addr", addr)
|
||||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||||
errCh <- err
|
errCh <- err
|
||||||
@@ -72,3 +80,9 @@ func RunPeriodic(ctx context.Context, log *slog.Logger, name string, interval ti
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TLSConfig enables HTTPS for the coordinator server.
|
||||||
|
type TLSConfig struct {
|
||||||
|
CertFile string
|
||||||
|
KeyFile string
|
||||||
|
}
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
_ "modernc.org/sqlite"
|
_ "modernc.org/sqlite"
|
||||||
@@ -37,9 +38,29 @@ func Open(path string) (*sql.DB, error) {
|
|||||||
_ = db.Close()
|
_ = db.Close()
|
||||||
return nil, fmt.Errorf("ping sqlite database: %w", err)
|
return nil, fmt.Errorf("ping sqlite database: %w", err)
|
||||||
}
|
}
|
||||||
|
if err := lockDownDatabase(path); err != nil {
|
||||||
|
_ = db.Close()
|
||||||
|
return nil, fmt.Errorf("lock down sqlite database: %w", err)
|
||||||
|
}
|
||||||
return db, nil
|
return db, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// lockDownDatabase restricts the database files to the owner: sqlite creates
|
||||||
|
// them with the process umask (0644), which would let any local user read job
|
||||||
|
// metadata and password hashes. WAL/SHM siblings inherit the main file's mode,
|
||||||
|
// so existing ones are corrected too. Best-effort: failures only warn callers
|
||||||
|
// via the returned error, never corrupt state.
|
||||||
|
func lockDownDatabase(path string) error {
|
||||||
|
for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
|
||||||
|
if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
|
||||||
|
if err := os.Chmod(candidate, 0o600); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// querier is satisfied by both *sql.DB and *sql.Tx, letting every repository
|
// querier is satisfied by both *sql.DB and *sql.Tx, letting every repository
|
||||||
// method run identically inside or outside a transaction.
|
// method run identically inside or outside a transaction.
|
||||||
type querier interface {
|
type querier interface {
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -347,3 +348,19 @@ func TestCancelByJobInvalidatesTasks(t *testing.T) {
|
|||||||
t.Errorf("cancelled task = %+v", got)
|
t.Errorf("cancelled task = %+v", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestOpenRestrictsDatabasePermissions(t *testing.T) {
|
||||||
|
path := filepath.Join(t.TempDir(), "locked.db")
|
||||||
|
db, err := Open(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
_ = db.Close()
|
||||||
|
info, err := os.Stat(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if perm := info.Mode().Perm(); perm != 0o600 {
|
||||||
|
t.Errorf("db perms = %o, want 600", perm)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,117 @@
|
|||||||
|
package http
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// documentedLimits describes the default policy for the two public surfaces;
|
||||||
|
// keep in sync with loginRatePerMinute and exchangeRatePerMinute below.
|
||||||
|
const (
|
||||||
|
loginRatePerMinute = 10
|
||||||
|
loginBurst = 5
|
||||||
|
exchangeRatePerMinute = 30
|
||||||
|
exchangeBurst = 10
|
||||||
|
)
|
||||||
|
|
||||||
|
// tokenBucket is a fixed-rate token bucket for one client address.
|
||||||
|
type tokenBucket struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
tokens float64
|
||||||
|
last time.Time
|
||||||
|
rate float64 // tokens per second
|
||||||
|
burst float64
|
||||||
|
}
|
||||||
|
|
||||||
|
func newTokenBucket(ratePerMinute, burst float64) *tokenBucket {
|
||||||
|
return &tokenBucket{
|
||||||
|
tokens: burst,
|
||||||
|
last: time.Now(),
|
||||||
|
rate: ratePerMinute / 60,
|
||||||
|
burst: burst,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// allow consumes one token when available; the bucket refills continuously.
|
||||||
|
func (b *tokenBucket) allow() bool {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
now := time.Now()
|
||||||
|
b.tokens += now.Sub(b.last).Seconds() * b.rate
|
||||||
|
if b.tokens > b.burst {
|
||||||
|
b.tokens = b.burst
|
||||||
|
}
|
||||||
|
b.last = now
|
||||||
|
if b.tokens < 1 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
b.tokens--
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// ipLimiter tracks one bucket per client address and prunes stale entries.
|
||||||
|
type ipLimiter struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
buckets map[string]*tokenBucket
|
||||||
|
ratePerMinute float64
|
||||||
|
burst float64
|
||||||
|
}
|
||||||
|
|
||||||
|
func newIPLimiter(ratePerMinute, burst float64) *ipLimiter {
|
||||||
|
return &ipLimiter{
|
||||||
|
buckets: make(map[string]*tokenBucket),
|
||||||
|
ratePerMinute: ratePerMinute,
|
||||||
|
burst: burst,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Allow reports whether the caller's address may proceed. It also sweeps
|
||||||
|
// entries idle for more than ten minutes so the map stays bounded.
|
||||||
|
func (l *ipLimiter) Allow(r *http.Request) bool {
|
||||||
|
ip := remoteIP(r)
|
||||||
|
l.mu.Lock()
|
||||||
|
if len(l.buckets) > 1024 {
|
||||||
|
cutoff := time.Now().Add(-10 * time.Minute)
|
||||||
|
for addr, bucket := range l.buckets {
|
||||||
|
bucket.mu.Lock()
|
||||||
|
idle := bucket.last.Before(cutoff)
|
||||||
|
bucket.mu.Unlock()
|
||||||
|
if idle {
|
||||||
|
delete(l.buckets, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
bucket, ok := l.buckets[ip]
|
||||||
|
if !ok {
|
||||||
|
bucket = newTokenBucket(l.ratePerMinute, l.burst)
|
||||||
|
l.buckets[ip] = bucket
|
||||||
|
}
|
||||||
|
l.mu.Unlock()
|
||||||
|
return bucket.allow()
|
||||||
|
}
|
||||||
|
|
||||||
|
func remoteIP(r *http.Request) string {
|
||||||
|
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||||
|
if err != nil {
|
||||||
|
return r.RemoteAddr
|
||||||
|
}
|
||||||
|
return host
|
||||||
|
}
|
||||||
|
|
||||||
|
// rateLimited wraps a handler with a per-address limiter; exhausted callers
|
||||||
|
// receive 429 with a Retry-After header.
|
||||||
|
func rateLimited(limiter *ipLimiter, next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if !limiter.Allow(r) {
|
||||||
|
w.Header().Set("Retry-After", "60")
|
||||||
|
writeJSON(w, http.StatusTooManyRequests, map[string]string{
|
||||||
|
"error": "too many requests, try again shortly",
|
||||||
|
"request_id": requestIDFrom(r.Context()),
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -0,0 +1,54 @@
|
|||||||
|
package http
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestTokenBucketBurstThenThrottles(t *testing.T) {
|
||||||
|
bucket := newTokenBucket(10, 3)
|
||||||
|
for i := 0; i < 3; i++ {
|
||||||
|
if !bucket.allow() {
|
||||||
|
t.Fatalf("request %d must pass within the burst", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if bucket.allow() {
|
||||||
|
t.Error("fourth request within the burst must be throttled")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRateLimitedReturns429(t *testing.T) {
|
||||||
|
limiter := newIPLimiter(10, 2)
|
||||||
|
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
})
|
||||||
|
handler := rateLimited(limiter, next)
|
||||||
|
|
||||||
|
req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/ui/login", nil)
|
||||||
|
req.RemoteAddr = "10.0.0.5:5555"
|
||||||
|
// Burst is 2: the first two pass, the third is throttled.
|
||||||
|
for i := 0; i < 2; i++ {
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("request %d: got %d", i, rec.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusTooManyRequests {
|
||||||
|
t.Fatalf("third request: got %d, want 429", rec.Code)
|
||||||
|
}
|
||||||
|
if rec.Header().Get("Retry-After") == "" {
|
||||||
|
t.Error("429 must carry Retry-After")
|
||||||
|
}
|
||||||
|
// A different address is not throttled by the same bucket.
|
||||||
|
other := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/ui/login", nil)
|
||||||
|
other.RemoteAddr = "10.0.0.6:5555"
|
||||||
|
rec = httptest.NewRecorder()
|
||||||
|
handler.ServeHTTP(rec, other)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Errorf("other client: got %d, want 200", rec.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -51,6 +51,8 @@ type Server struct {
|
|||||||
// userserviceURL is the base URL the UI proxies login/registration to. Empty
|
// userserviceURL is the base URL the UI proxies login/registration to. Empty
|
||||||
// keeps the static basic-auth UI.
|
// keeps the static basic-auth UI.
|
||||||
userserviceURL string
|
userserviceURL string
|
||||||
|
// disableRegistration forbids new accounts; login keeps working.
|
||||||
|
disableRegistration bool
|
||||||
// publicCoordinatorURL / publicUserserviceURL are the browser-facing URLs
|
// publicCoordinatorURL / publicUserserviceURL are the browser-facing URLs
|
||||||
// rendered into the worker-enrollment command. Either may be empty; the
|
// rendered into the worker-enrollment command. Either may be empty; the
|
||||||
// template falls back (own origin / userserviceURL respectively).
|
// template falls back (own origin / userserviceURL respectively).
|
||||||
@@ -70,6 +72,19 @@ type Server struct {
|
|||||||
func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
|
func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
|
||||||
maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error,
|
maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error,
|
||||||
publicURLs ...string) *Server {
|
publicURLs ...string) *Server {
|
||||||
|
return NewServerWithOptions(uc, log, requestTimeout, heartbeatInterval, maxUploadBytes, jwtSecret, userserviceURL, m, ready, ServerOptions{}, publicURLs...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ServerOptions configures non-positional behaviour of the operator UI.
|
||||||
|
type ServerOptions struct {
|
||||||
|
DisableRegistration bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewServerWithOptions is NewServer plus explicit options; the option-less
|
||||||
|
// variant exists so existing call sites and tests need no change.
|
||||||
|
func NewServerWithOptions(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
|
||||||
|
maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error,
|
||||||
|
opts ServerOptions, publicURLs ...string) *Server {
|
||||||
if m == nil {
|
if m == nil {
|
||||||
m = metrics.New()
|
m = metrics.New()
|
||||||
}
|
}
|
||||||
@@ -88,6 +103,7 @@ func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval
|
|||||||
}
|
}
|
||||||
return &Server{
|
return &Server{
|
||||||
uc: uc,
|
uc: uc,
|
||||||
|
disableRegistration: opts.DisableRegistration,
|
||||||
log: log,
|
log: log,
|
||||||
requestTimeout: requestTimeout,
|
requestTimeout: requestTimeout,
|
||||||
heartbeatInterval: heartbeatInterval,
|
heartbeatInterval: heartbeatInterval,
|
||||||
@@ -135,7 +151,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
|
|||||||
// Worker-key exchange is fronted by the coordinator when the userservice
|
// Worker-key exchange is fronted by the coordinator when the userservice
|
||||||
// is embedded (serve mode): the key itself is the credential.
|
// is embedded (serve mode): the key itself is the credential.
|
||||||
if s.userserviceURL != "" {
|
if s.userserviceURL != "" {
|
||||||
mux.HandleFunc("POST /worker-tokens/exchange", s.handleWorkerTokenExchangeProxy)
|
mux.Handle("POST /worker-tokens/exchange", rateLimited(newIPLimiter(exchangeRatePerMinute, exchangeBurst), http.HandlerFunc(s.handleWorkerTokenExchangeProxy)))
|
||||||
}
|
}
|
||||||
|
|
||||||
hasBasicAuth := len(uiToken) > 0 && uiToken[0] != ""
|
hasBasicAuth := len(uiToken) > 0 && uiToken[0] != ""
|
||||||
@@ -163,7 +179,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
|
|||||||
if s.uiSessionMode() {
|
if s.uiSessionMode() {
|
||||||
// Public auth pages — reachable without a session so a user can log in.
|
// Public auth pages — reachable without a session so a user can log in.
|
||||||
ui.HandleFunc("GET /ui/login", s.handleUILoginForm)
|
ui.HandleFunc("GET /ui/login", s.handleUILoginForm)
|
||||||
ui.HandleFunc("POST /ui/login", s.handleUILogin)
|
ui.Handle("POST /ui/login", rateLimited(newIPLimiter(loginRatePerMinute, loginBurst), http.HandlerFunc(s.handleUILogin)))
|
||||||
ui.HandleFunc("GET /ui/logout-form", s.handleUILogoutForm)
|
ui.HandleFunc("GET /ui/logout-form", s.handleUILogoutForm)
|
||||||
ui.HandleFunc("GET /ui/register", s.handleUIRegisterForm)
|
ui.HandleFunc("GET /ui/register", s.handleUIRegisterForm)
|
||||||
ui.HandleFunc("POST /ui/register", s.handleUIRegister)
|
ui.HandleFunc("POST /ui/register", s.handleUIRegister)
|
||||||
|
|||||||
@@ -108,6 +108,10 @@ func (s *Server) handleUILogin(w http.ResponseWriter, r *http.Request) {
|
|||||||
// user to the login page. The new account is a plain user until an admin
|
// user to the login page. The new account is a plain user until an admin
|
||||||
// promotes or verifies it.
|
// promotes or verifies it.
|
||||||
func (s *Server) handleUIRegister(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleUIRegister(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if s.disableRegistration {
|
||||||
|
http.Redirect(w, r, "/ui/register?error=registration+disabled", http.StatusSeeOther)
|
||||||
|
return
|
||||||
|
}
|
||||||
email, password := r.FormValue("email"), r.FormValue("password")
|
email, password := r.FormValue("email"), r.FormValue("password")
|
||||||
|
|
||||||
status, _, err := s.callUserservice(r.Context(), "/register", email, password)
|
status, _, err := s.callUserservice(r.Context(), "/register", email, password)
|
||||||
|
|||||||
@@ -226,3 +226,18 @@ func TestLoginFormRendersNext(t *testing.T) {
|
|||||||
t.Error("login form must not render next when absent")
|
t.Error("login form must not render next when absent")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRegistrationDisabledRejectsNewAccounts(t *testing.T) {
|
||||||
|
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
t.Fatal("userservice must not be called when registration is disabled")
|
||||||
|
}))
|
||||||
|
defer stub.Close()
|
||||||
|
s := newLoginServer(stub)
|
||||||
|
s.disableRegistration = true
|
||||||
|
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
s.handleUIRegister(rec, postForm("/ui/register", url.Values{"email": {"a@b.io"}, "password": {"pw"}}))
|
||||||
|
if rec.Code != http.StatusSeeOther || !strings.Contains(rec.Header().Get("Location"), "registration+disabled") {
|
||||||
|
t.Errorf("got %d -> %q, want 303 to the registration-disabled error", rec.Code, rec.Header().Get("Location"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
_ "modernc.org/sqlite"
|
_ "modernc.org/sqlite"
|
||||||
@@ -290,6 +291,17 @@ func indexOf(haystack, needle string) int {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Open opens (and creates when missing) the userservice database file.
|
// Open opens (and creates when missing) the userservice database file.
|
||||||
|
func lockDownDatabase(path string) error {
|
||||||
|
for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
|
||||||
|
if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
|
||||||
|
if err := os.Chmod(candidate, 0o600); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func Open(path string) (*sql.DB, error) {
|
func Open(path string) (*sql.DB, error) {
|
||||||
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=synchronous(NORMAL)", path)
|
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=synchronous(NORMAL)", path)
|
||||||
db, err := sql.Open("sqlite", dsn)
|
db, err := sql.Open("sqlite", dsn)
|
||||||
@@ -300,5 +312,9 @@ func Open(path string) (*sql.DB, error) {
|
|||||||
_ = db.Close()
|
_ = db.Close()
|
||||||
return nil, fmt.Errorf("ping userservice database: %w", err)
|
return nil, fmt.Errorf("ping userservice database: %w", err)
|
||||||
}
|
}
|
||||||
|
if err := lockDownDatabase(path); err != nil {
|
||||||
|
_ = db.Close()
|
||||||
|
return nil, fmt.Errorf("lock down userservice database: %w", err)
|
||||||
|
}
|
||||||
return db, nil
|
return db, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"os"
|
||||||
|
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
|
|
||||||
@@ -38,6 +39,12 @@ func (h *Handlers) handleHealth(w http.ResponseWriter, _ *http.Request) {
|
|||||||
// handleRegister creates an account. It returns 201 with the public user view,
|
// handleRegister creates an account. It returns 201 with the public user view,
|
||||||
// 409 if the email is taken, or 400 on a malformed body / weak password.
|
// 409 if the email is taken, or 400 on a malformed body / weak password.
|
||||||
func (h *Handlers) handleRegister(w http.ResponseWriter, r *http.Request) {
|
func (h *Handlers) handleRegister(w http.ResponseWriter, r *http.Request) {
|
||||||
|
// Standalone deployments can close self-service registration while keeping
|
||||||
|
// the bootstrap admin and existing accounts (USERSERVICE_DISABLE_REGISTRATION=1).
|
||||||
|
if os.Getenv("USERSERVICE_DISABLE_REGISTRATION") == "1" {
|
||||||
|
writeJSON(w, http.StatusForbidden, errorResponse{Error: "registration disabled", RequestID: requestIDFrom(r.Context())})
|
||||||
|
return
|
||||||
|
}
|
||||||
var req registerRequest
|
var req registerRequest
|
||||||
if !decodeJSON(w, r, &req) {
|
if !decodeJSON(w, r, &req) {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -0,0 +1,97 @@
|
|||||||
|
package http
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// The same per-address token-bucket policy as the coordinator transport:
|
||||||
|
// login is the credential brute-force surface, the exchange the only public
|
||||||
|
// token-minting one.
|
||||||
|
const (
|
||||||
|
loginRatePerMinute = 10
|
||||||
|
loginBurst = 5
|
||||||
|
exchangeRatePerMinute = 30
|
||||||
|
exchangeBurst = 10
|
||||||
|
)
|
||||||
|
|
||||||
|
type tokenBucket struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
tokens float64
|
||||||
|
last time.Time
|
||||||
|
rate float64
|
||||||
|
burst float64
|
||||||
|
}
|
||||||
|
|
||||||
|
func newTokenBucket(ratePerMinute, burst float64) *tokenBucket {
|
||||||
|
return &tokenBucket{tokens: burst, last: time.Now(), rate: ratePerMinute / 60, burst: burst}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *tokenBucket) allow() bool {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
now := time.Now()
|
||||||
|
b.tokens += now.Sub(b.last).Seconds() * b.rate
|
||||||
|
if b.tokens > b.burst {
|
||||||
|
b.tokens = b.burst
|
||||||
|
}
|
||||||
|
b.last = now
|
||||||
|
if b.tokens < 1 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
b.tokens--
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
type ipLimiter struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
buckets map[string]*tokenBucket
|
||||||
|
ratePerMinute float64
|
||||||
|
burst float64
|
||||||
|
}
|
||||||
|
|
||||||
|
func newIPLimiter(ratePerMinute, burst float64) *ipLimiter {
|
||||||
|
return &ipLimiter{buckets: map[string]*tokenBucket{}, ratePerMinute: ratePerMinute, burst: burst}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *ipLimiter) Allow(r *http.Request) bool {
|
||||||
|
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||||
|
if err != nil {
|
||||||
|
host = r.RemoteAddr
|
||||||
|
}
|
||||||
|
l.mu.Lock()
|
||||||
|
if len(l.buckets) > 1024 {
|
||||||
|
cutoff := time.Now().Add(-10 * time.Minute)
|
||||||
|
for addr, bucket := range l.buckets {
|
||||||
|
bucket.mu.Lock()
|
||||||
|
idle := bucket.last.Before(cutoff)
|
||||||
|
bucket.mu.Unlock()
|
||||||
|
if idle {
|
||||||
|
delete(l.buckets, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
bucket, ok := l.buckets[host]
|
||||||
|
if !ok {
|
||||||
|
bucket = newTokenBucket(l.ratePerMinute, l.burst)
|
||||||
|
l.buckets[host] = bucket
|
||||||
|
}
|
||||||
|
l.mu.Unlock()
|
||||||
|
return bucket.allow()
|
||||||
|
}
|
||||||
|
|
||||||
|
func rateLimited(limiter *ipLimiter, next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if !limiter.Allow(r) {
|
||||||
|
w.Header().Set("Retry-After", "60")
|
||||||
|
writeJSON(w, http.StatusTooManyRequests, errorResponse{
|
||||||
|
Error: "too many requests, try again shortly",
|
||||||
|
RequestID: requestIDFrom(r.Context()),
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -51,14 +51,14 @@ func NewServer(log *slog.Logger, uc UseCases, issuer auth.Issuer) http.Handler {
|
|||||||
// Method-aware patterns (Go 1.22+): a GET to /register is a 405, not a match.
|
// Method-aware patterns (Go 1.22+): a GET to /register is a 405, not a match.
|
||||||
mux.HandleFunc("GET /health", h.handleHealth)
|
mux.HandleFunc("GET /health", h.handleHealth)
|
||||||
mux.HandleFunc("POST /register", h.handleRegister)
|
mux.HandleFunc("POST /register", h.handleRegister)
|
||||||
mux.HandleFunc("POST /login", h.handleLogin)
|
mux.Handle("POST /login", rateLimited(newIPLimiter(loginRatePerMinute, loginBurst), http.HandlerFunc(h.handleLogin)))
|
||||||
// /me proves a token round-trips; it sits behind JWT auth.
|
// /me proves a token round-trips; it sits behind JWT auth.
|
||||||
mux.Handle("GET /me", chain(http.HandlerFunc(h.handleMe), withJWT(issuer)))
|
mux.Handle("GET /me", chain(http.HandlerFunc(h.handleMe), withJWT(issuer)))
|
||||||
|
|
||||||
// Worker keys: a user mints a long-lived key (JWT-protected), and a worker
|
// Worker keys: a user mints a long-lived key (JWT-protected), and a worker
|
||||||
// trades it for a short-lived JWT on the public exchange endpoint — the key
|
// trades it for a short-lived JWT on the public exchange endpoint — the key
|
||||||
// itself is the credential there, so no prior token is required.
|
// itself is the credential there, so no prior token is required.
|
||||||
mux.HandleFunc("POST /worker-tokens/exchange", h.handleExchangeWorkerKey)
|
mux.Handle("POST /worker-tokens/exchange", rateLimited(newIPLimiter(exchangeRatePerMinute, exchangeBurst), http.HandlerFunc(h.handleExchangeWorkerKey)))
|
||||||
mux.Handle("POST /worker-keys", chain(http.HandlerFunc(h.handleCreateWorkerKey), withJWT(issuer)))
|
mux.Handle("POST /worker-keys", chain(http.HandlerFunc(h.handleCreateWorkerKey), withJWT(issuer)))
|
||||||
mux.Handle("GET /worker-keys", chain(http.HandlerFunc(h.handleListWorkerKeys), withJWT(issuer)))
|
mux.Handle("GET /worker-keys", chain(http.HandlerFunc(h.handleListWorkerKeys), withJWT(issuer)))
|
||||||
mux.Handle("DELETE /worker-keys/{id}", chain(http.HandlerFunc(h.handleRevokeWorkerKey), withJWT(issuer)))
|
mux.Handle("DELETE /worker-keys/{id}", chain(http.HandlerFunc(h.handleRevokeWorkerKey), withJWT(issuer)))
|
||||||
|
|||||||
@@ -479,3 +479,12 @@ func TestAdminListsUsersAndKeys(t *testing.T) {
|
|||||||
t.Errorf("admin revoke unknown key: got %d, want 404", rec.Code)
|
t.Errorf("admin revoke unknown key: got %d, want 404", rec.Code)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRegistrationDisabledEnv(t *testing.T) {
|
||||||
|
t.Setenv("USERSERVICE_DISABLE_REGISTRATION", "1")
|
||||||
|
h := newTestServer()
|
||||||
|
rec := do(t, h, http.MethodPost, "/register", "", map[string]string{"email": "blocked@x.io", "password": "pw"})
|
||||||
|
if rec.Code != http.StatusForbidden {
|
||||||
|
t.Errorf("register when disabled: got %d, want 403", rec.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+62
@@ -8,6 +8,11 @@
|
|||||||
# coordinator serve --open
|
# coordinator serve --open
|
||||||
$ErrorActionPreference = "Stop"
|
$ErrorActionPreference = "Stop"
|
||||||
|
|
||||||
|
# Public half of the Ed25519 key that signs SHA256SUMS.txt in releases (see
|
||||||
|
# install.sh). Verification needs the openssl binary; without it the installer
|
||||||
|
# falls back to checksum verification with a warning.
|
||||||
|
$ScimeshSigningPubKey = "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA01rjmCme4W4zAgBwbO00LvwgnB1srlg0LbooRG8ej7iNxzOtJ8vjRFR2Cu7z7OKjoDo9/0GW3pvcwB+ndBB6yUwht33IRwdsnbioBI4M7LL+yC1ubi4fJ5bigOgZ9VsVqKdU3T9GYxmrfJF1UexiOg6HjoRLO3V4Id+3e/CiI5Sr8UMfJMXUfO3uiEs9RpstxpP1V/UU4YDicTF0QjkOESimEwwXBG4z3VcVmQtqkb7Q3413iekTdQ13093GKAKp0Q2ia1TpB2su6ELUhHAqhmK88cJ73Opy1uEVye0twov4BFTu5GkxgazNTuU//aYVWVpd/NAlD+VVSmpDsbfBBQIDAQAB"
|
||||||
|
|
||||||
$Repo = "emil28092005/SciMesh"
|
$Repo = "emil28092005/SciMesh"
|
||||||
$Component = if ($env:SCIMESH_COMPONENT) { $env:SCIMESH_COMPONENT } else { "coordinator" }
|
$Component = if ($env:SCIMESH_COMPONENT) { $env:SCIMESH_COMPONENT } else { "coordinator" }
|
||||||
$Version = if ($env:SCIMESH_VERSION) { $env:SCIMESH_VERSION } else { "latest" }
|
$Version = if ($env:SCIMESH_VERSION) { $env:SCIMESH_VERSION } else { "latest" }
|
||||||
@@ -57,6 +62,63 @@ $Target = Join-Path $InstallDir "$Binary.exe"
|
|||||||
|
|
||||||
Write-Host "Downloading $Url"
|
Write-Host "Downloading $Url"
|
||||||
Invoke-WebRequest -Uri $Url -OutFile "$Target.tmp"
|
Invoke-WebRequest -Uri $Url -OutFile "$Target.tmp"
|
||||||
|
|
||||||
|
# Verify the SHA-256 checksum from the release before installing (see
|
||||||
|
# install.sh for the caveats). $env:SCIMESH_SKIP_VERIFY -eq "1" bypasses.
|
||||||
|
if ($env:SCIMESH_SKIP_VERIFY -ne "1") {
|
||||||
|
try {
|
||||||
|
$SumUrl = "https://github.com/$Repo/releases/download/$Version/SHA256SUMS.txt"
|
||||||
|
|
||||||
|
# Ed25519 signature over the checksum file, when openssl is present.
|
||||||
|
# Both files are fetched with -OutFile so their bytes match the
|
||||||
|
# release exactly (string pipelines would rewrite line endings).
|
||||||
|
$openssl = Get-Command openssl -ErrorAction SilentlyContinue
|
||||||
|
if ($env:SCIMESH_SKIP_SIGNATURE -ne "1" -and $openssl) {
|
||||||
|
$PubFile = Join-Path $env:TEMP "scimesh-signing-pub.pem"
|
||||||
|
$SumFile = Join-Path $env:TEMP ("scimesh-sums-" + [guid]::NewGuid().ToString("N") + ".txt")
|
||||||
|
$SigFile = "$SumFile.sig"
|
||||||
|
Set-Content -Path $PubFile -Value @("-----BEGIN PUBLIC KEY-----", $ScimeshSigningPubKey, "-----END PUBLIC KEY-----")
|
||||||
|
try {
|
||||||
|
Invoke-WebRequest -Uri $SumUrl -OutFile $SumFile
|
||||||
|
Invoke-WebRequest -Uri "$SumUrl.sig" -OutFile $SigFile
|
||||||
|
& $openssl.Source dgst -sha256 -verify $PubFile -signature $SigFile $SumFile 2>&1 | Out-Null
|
||||||
|
if ($LASTEXITCODE -eq 0) {
|
||||||
|
Write-Host "Signature verified (RSA-2048/SHA-256)"
|
||||||
|
} else {
|
||||||
|
Remove-Item -Force "$Target.tmp"
|
||||||
|
throw "the release signature does not verify; the download channel may be tampered with"
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
Remove-Item -Force "$Target.tmp"
|
||||||
|
throw "signature verification failed: $($_.Exception.Message)"
|
||||||
|
} finally {
|
||||||
|
Remove-Item -Force $PubFile, $SumFile, $SigFile -ErrorAction SilentlyContinue
|
||||||
|
}
|
||||||
|
} elseif ($env:SCIMESH_SKIP_SIGNATURE -ne "1") {
|
||||||
|
Write-Host "WARNING: openssl not found; falling back to checksum verification only"
|
||||||
|
}
|
||||||
|
|
||||||
|
$BinaryName = Split-Path $Url -Leaf
|
||||||
|
$SumFileCheck = Join-Path $env:TEMP ("scimesh-sums-check-" + [guid]::NewGuid().ToString("N") + ".txt")
|
||||||
|
Invoke-WebRequest -Uri $SumUrl -OutFile $SumFileCheck
|
||||||
|
$Line = (Get-Content $SumFileCheck -Raw -ErrorAction SilentlyContinue -split "`n") | Where-Object { $_.Trim().EndsWith(" " + $BinaryName) } | Select-Object -First 1
|
||||||
|
if ($Line) {
|
||||||
|
$Expected = ($Line -split "\s+")[0]
|
||||||
|
$Actual = (Get-FileHash -Algorithm SHA256 -Path "$Target.tmp").Hash.ToLower()
|
||||||
|
if ($Actual -ne $Expected.ToLower()) {
|
||||||
|
Remove-Item -Force "$Target.tmp", $SumFileCheck
|
||||||
|
throw "checksum mismatch for $Binary (got $Actual, want $Expected)"
|
||||||
|
}
|
||||||
|
Write-Host "Checksum verified ($($Expected.Substring(0,12))...)"
|
||||||
|
} else {
|
||||||
|
Write-Host "WARNING: no checksum entry for $Binary; skipping verification"
|
||||||
|
Remove-Item -Force $SumFileCheck -ErrorAction SilentlyContinue
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
Write-Host "WARNING: could not verify checksum ($($_.Exception.Message)); continuing"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Move-Item -Force "$Target.tmp" $Target
|
Move-Item -Force "$Target.tmp" $Target
|
||||||
|
|
||||||
Write-Host ""
|
Write-Host ""
|
||||||
|
|||||||
+51
@@ -14,6 +14,13 @@
|
|||||||
set -eu
|
set -eu
|
||||||
|
|
||||||
REPO="emil28092005/SciMesh"
|
REPO="emil28092005/SciMesh"
|
||||||
|
|
||||||
|
# Public half of the Ed25519 key that signs SHA256SUMS.txt in releases. The
|
||||||
|
# private half lives in the repository secret SCIMESH_SIGNING_KEY. Verification
|
||||||
|
# uses openssl when available; without openssl the installer falls back to the
|
||||||
|
# checksum-only check with a warning.
|
||||||
|
SCIMESH_SIGNING_PUBKEY='MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA01rjmCme4W4zAgBwbO00LvwgnB1srlg0LbooRG8ej7iNxzOtJ8vjRFR2Cu7z7OKjoDo9/0GW3pvcwB+ndBB6yUwht33IRwdsnbioBI4M7LL+yC1ubi4fJ5bigOgZ9VsVqKdU3T9GYxmrfJF1UexiOg6HjoRLO3V4Id+3e/CiI5Sr8UMfJMXUfO3uiEs9RpstxpP1V/UU4YDicTF0QjkOESimEwwXBG4z3VcVmQtqkb7Q3413iekTdQ13093GKAKp0Q2ia1TpB2su6ELUhHAqhmK88cJ73Opy1uEVye0twov4BFTu5GkxgazNTuU//aYVWVpd/NAlD+VVSmpDsbfBBQIDAQAB'
|
||||||
|
|
||||||
COMPONENT="${1:-coordinator}"
|
COMPONENT="${1:-coordinator}"
|
||||||
VERSION="${SCIMESH_VERSION:-latest}"
|
VERSION="${SCIMESH_VERSION:-latest}"
|
||||||
INSTALL_DIR="${SCIMESH_INSTALL_DIR:-$HOME/.local/bin}"
|
INSTALL_DIR="${SCIMESH_INSTALL_DIR:-$HOME/.local/bin}"
|
||||||
@@ -60,6 +67,50 @@ TARGET="$INSTALL_DIR/$BINARY"
|
|||||||
URL="https://github.com/${REPO}/releases/download/${VERSION}/${BINARY}-${OS}-${ARCH}"
|
URL="https://github.com/${REPO}/releases/download/${VERSION}/${BINARY}-${OS}-${ARCH}"
|
||||||
echo "Downloading $URL"
|
echo "Downloading $URL"
|
||||||
curl -fsSL -o "$TARGET.tmp" "$URL"
|
curl -fsSL -o "$TARGET.tmp" "$URL"
|
||||||
|
|
||||||
|
# Verify the SHA-256 checksum from the release before installing. This guards
|
||||||
|
# against corrupted downloads and stale CDN caches; it does not protect
|
||||||
|
# against an active MITM on the same channel (the checksum file travels it
|
||||||
|
# too). Set SCIMESH_SKIP_VERIFY=1 to bypass.
|
||||||
|
if [ "${SCIMESH_SKIP_VERIFY:-0}" != "1" ]; then
|
||||||
|
if SUMFILE=$(mktemp) && curl -fsSL -o "$SUMFILE" "https://github.com/${REPO}/releases/download/${VERSION}/SHA256SUMS.txt"; then
|
||||||
|
EXPECTED=$(awk '$2 == "'"$(basename "$URL")"'" {print $1}' "$SUMFILE" 2>/dev/null | head -1)
|
||||||
|
SIGFILE="$SUMFILE.sig"
|
||||||
|
if [ "${SCIMESH_SKIP_SIGNATURE:-0}" != "1" ] && command -v openssl >/dev/null 2>&1 \
|
||||||
|
&& curl -fsSL -o "$SIGFILE" "https://github.com/${REPO}/releases/download/${VERSION}/SHA256SUMS.txt.sig" 2>/dev/null; then
|
||||||
|
PUBKEY_FILE=$(mktemp)
|
||||||
|
printf '%s\n' '-----BEGIN PUBLIC KEY-----' "$SCIMESH_SIGNING_PUBKEY" '-----END PUBLIC KEY-----' > "$PUBKEY_FILE"
|
||||||
|
if openssl dgst -sha256 -verify "$PUBKEY_FILE" -signature "$SIGFILE" "$SUMFILE" >/dev/null 2>&1; then
|
||||||
|
echo "Signature verified (RSA-2048/SHA-256)"
|
||||||
|
else
|
||||||
|
rm -f "$PUBKEY_FILE" "$SIGFILE" "$SUMFILE" "$TARGET.tmp"
|
||||||
|
echo "ERROR: the release signature does not verify; the download channel may be tampered with." >&2
|
||||||
|
echo "Retry later, or bypass with SCIMESH_SKIP_SIGNATURE=1." >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
rm -f "$PUBKEY_FILE"
|
||||||
|
elif [ "${SCIMESH_SKIP_SIGNATURE:-0}" != "1" ] && ! command -v openssl >/dev/null 2>&1; then
|
||||||
|
echo "WARNING: openssl not found; falling back to checksum verification only"
|
||||||
|
fi
|
||||||
|
rm -f "$SUMFILE" "$SIGFILE"
|
||||||
|
if [ -n "$EXPECTED" ]; then
|
||||||
|
ACTUAL=$(sha256sum "$TARGET.tmp" | awk '{print $1}')
|
||||||
|
if [ "$ACTUAL" != "$EXPECTED" ]; then
|
||||||
|
rm -f "$TARGET.tmp"
|
||||||
|
echo "ERROR: checksum mismatch for $BINARY (got $ACTUAL, want $EXPECTED)" >&2
|
||||||
|
echo "The download may be corrupted or served by a stale cache. Retry later, or" >&2
|
||||||
|
echo "pin the version with SCIMESH_VERSION=${VERSION} and re-run." >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
echo "Checksum verified ($(echo "$EXPECTED" | cut -c1-12)…)"
|
||||||
|
else
|
||||||
|
echo "WARNING: no checksum entry for $(basename "$URL"); skipping verification"
|
||||||
|
fi
|
||||||
|
else
|
||||||
|
echo "WARNING: could not fetch SHA256SUMS.txt; skipping verification"
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
|
||||||
chmod +x "$TARGET.tmp"
|
chmod +x "$TARGET.tmp"
|
||||||
mv "$TARGET.tmp" "$TARGET"
|
mv "$TARGET.tmp" "$TARGET"
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
# SciMesh uninstaller for Windows: stops the running component, removes its
|
||||||
|
# binary, and — when asked — deletes its data directory.
|
||||||
|
#
|
||||||
|
# powershell -ExecutionPolicy Bypass -c "irm https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.ps1 | iex"
|
||||||
|
#
|
||||||
|
# $env:SCIMESH_COMPONENT selects the component (coordinator | worker | all,
|
||||||
|
# default all). Data is kept unless $env:SCIMESH_PURGE -eq "1" or the
|
||||||
|
# -Purge switch is passed (in a pipe there is no interactive prompt).
|
||||||
|
param(
|
||||||
|
[switch]$Purge
|
||||||
|
)
|
||||||
|
$ErrorActionPreference = "Stop"
|
||||||
|
|
||||||
|
$Component = if ($env:SCIMESH_COMPONENT) { $env:SCIMESH_COMPONENT } else { "all" }
|
||||||
|
if ($Purge -or $env:SCIMESH_PURGE -eq "1") { $Purge = $true } else { $Purge = $false }
|
||||||
|
|
||||||
|
$InstallDir = if ($env:SCIMESH_INSTALL_DIR) {
|
||||||
|
$env:SCIMESH_INSTALL_DIR
|
||||||
|
} else {
|
||||||
|
Join-Path $env:LOCALAPPDATA "SciMesh"
|
||||||
|
}
|
||||||
|
|
||||||
|
function Remove-Component {
|
||||||
|
param([string]$Name, [string]$Binary)
|
||||||
|
Write-Host "Stopping $Name…"
|
||||||
|
Get-Process -Name $Name -ErrorAction SilentlyContinue | Stop-Process -Force -ErrorAction SilentlyContinue
|
||||||
|
$target = Join-Path $InstallDir $Binary
|
||||||
|
if (Test-Path $target) {
|
||||||
|
Write-Host "Removing $target…"
|
||||||
|
Remove-Item -Force $target
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function Remove-Data {
|
||||||
|
param([string]$Dir, [string]$Label)
|
||||||
|
if (-not (Test-Path $Dir)) { return }
|
||||||
|
if ($Purge) {
|
||||||
|
Remove-Item -Recurse -Force $Dir
|
||||||
|
Write-Host "Deleted $Dir"
|
||||||
|
return
|
||||||
|
}
|
||||||
|
# No interactive prompt in a pipe: keep the data by default.
|
||||||
|
Write-Host "Keeping $Dir ($Label). Pass -Purge to delete it."
|
||||||
|
}
|
||||||
|
|
||||||
|
switch ($Component) {
|
||||||
|
"coordinator" {
|
||||||
|
Remove-Component -Name "coordinator" -Binary "coordinator.exe"
|
||||||
|
Remove-Data -Dir (Join-Path $HOME ".scimesh") -Label "secrets, databases, artifacts, users"
|
||||||
|
}
|
||||||
|
"worker" {
|
||||||
|
Remove-Component -Name "worker-agent" -Binary "worker-agent.exe"
|
||||||
|
Remove-Data -Dir (Join-Path $HOME ".scimesh-worker") -Label "worker config, runtime venv, logs"
|
||||||
|
}
|
||||||
|
"all" {
|
||||||
|
Remove-Component -Name "coordinator" -Binary "coordinator.exe"
|
||||||
|
Remove-Component -Name "worker-agent" -Binary "worker-agent.exe"
|
||||||
|
Remove-Data -Dir (Join-Path $HOME ".scimesh") -Label "secrets, databases, artifacts, users"
|
||||||
|
Remove-Data -Dir (Join-Path $HOME ".scimesh-worker") -Label "worker config, runtime venv, logs"
|
||||||
|
}
|
||||||
|
default { throw "unknown component: $Component (use 'coordinator', 'worker' or 'all')" }
|
||||||
|
}
|
||||||
|
|
||||||
|
Write-Host ""
|
||||||
|
Write-Host "SciMesh $Component uninstalled."
|
||||||
|
Write-Host "Pass -Purge to also delete the data directories."
|
||||||
@@ -0,0 +1,83 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# SciMesh uninstaller: stops the running component, removes its binary, and —
|
||||||
|
# when asked — deletes its data directory.
|
||||||
|
#
|
||||||
|
# curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.sh | bash -s coordinator
|
||||||
|
# curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.sh | bash -s worker
|
||||||
|
# curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/uninstall.sh # both
|
||||||
|
#
|
||||||
|
# Data (jobs, artifacts, users, secrets, the managed venv) is kept by default.
|
||||||
|
# Pass --purge (or set SCIMESH_PURGE=1) to delete it without asking; without
|
||||||
|
# it the script prompts interactively. Because `curl | bash` pipes have no
|
||||||
|
# interactive stdin, the default is always "keep data".
|
||||||
|
set -eu
|
||||||
|
|
||||||
|
COMPONENT="${1:-all}"
|
||||||
|
PURGE=0
|
||||||
|
for arg in "$@"; do
|
||||||
|
case "$arg" in
|
||||||
|
--purge) PURGE=1 ;;
|
||||||
|
esac
|
||||||
|
done
|
||||||
|
[ "${SCIMESH_PURGE:-0}" = "1" ] && PURGE=1
|
||||||
|
|
||||||
|
INSTALL_DIR="${SCIMESH_INSTALL_DIR:-$HOME/.local/bin}"
|
||||||
|
|
||||||
|
remove_component() {
|
||||||
|
case "$1" in
|
||||||
|
coordinator)
|
||||||
|
echo "Stopping the coordinator (serve)…"
|
||||||
|
pkill -x coordinator 2>/dev/null || true
|
||||||
|
echo "Removing $INSTALL_DIR/coordinator…"
|
||||||
|
rm -f "$INSTALL_DIR/coordinator"
|
||||||
|
;;
|
||||||
|
worker)
|
||||||
|
echo "Stopping the worker agent and its setup wizard…"
|
||||||
|
pkill -x worker-agent 2>/dev/null || true
|
||||||
|
echo "Removing $INSTALL_DIR/worker-agent…"
|
||||||
|
rm -f "$INSTALL_DIR/worker-agent"
|
||||||
|
;;
|
||||||
|
*) echo "unknown component: $1 (use 'coordinator', 'worker' or 'all')" >&2; exit 1 ;;
|
||||||
|
esac
|
||||||
|
}
|
||||||
|
|
||||||
|
remove_data() {
|
||||||
|
local dir="$1" label="$2"
|
||||||
|
if [ ! -d "$dir" ]; then
|
||||||
|
return 0
|
||||||
|
fi
|
||||||
|
if [ "$PURGE" = "1" ]; then
|
||||||
|
rm -rf "$dir"
|
||||||
|
echo "Deleted $dir"
|
||||||
|
return 0
|
||||||
|
fi
|
||||||
|
# In a pipe (curl | bash) stdin is exhausted, so the prompt defaults to keep.
|
||||||
|
printf "Delete %s (%s)? [y/N] " "$dir" "$label"
|
||||||
|
read -r answer || answer=""
|
||||||
|
case "$answer" in
|
||||||
|
y|Y|yes|YES) rm -rf "$dir"; echo "Deleted $dir" ;;
|
||||||
|
*) echo "Keeping $dir" ;;
|
||||||
|
esac
|
||||||
|
}
|
||||||
|
|
||||||
|
case "$COMPONENT" in
|
||||||
|
coordinator)
|
||||||
|
remove_component coordinator
|
||||||
|
remove_data "$HOME/.scimesh" "secrets, databases, artifacts, users, managed venv"
|
||||||
|
;;
|
||||||
|
worker)
|
||||||
|
remove_component worker
|
||||||
|
remove_data "$HOME/.scimesh-worker" "worker config, runtime venv, logs"
|
||||||
|
;;
|
||||||
|
all)
|
||||||
|
remove_component coordinator
|
||||||
|
remove_component worker
|
||||||
|
remove_data "$HOME/.scimesh" "secrets, databases, artifacts, users, managed venv"
|
||||||
|
remove_data "$HOME/.scimesh-worker" "worker config, runtime venv, logs"
|
||||||
|
;;
|
||||||
|
*) echo "unknown component: $COMPONENT (use 'coordinator', 'worker' or 'all')" >&2; exit 1 ;;
|
||||||
|
esac
|
||||||
|
|
||||||
|
echo
|
||||||
|
echo "SciMesh $COMPONENT uninstalled."
|
||||||
|
echo "Pass --purge to also delete the data directories without asking."
|
||||||
Reference in New Issue
Block a user