Compare commits
7
Commits
v1.1.0-alpha.23
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
25fa6a488a | ||
|
|
78dacb7e17 | ||
|
|
c2c8336438 | ||
|
|
63c8ef0b8a | ||
|
|
049113cec8 | ||
|
|
330f95a375 | ||
|
|
e744e62d03 |
@@ -104,6 +104,20 @@ jobs:
|
||||
working-directory: artifacts
|
||||
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
|
||||
with:
|
||||
files: |
|
||||
|
||||
@@ -63,6 +63,14 @@ powershell -ExecutionPolicy Bypass -c "irm https://raw.githubusercontent.com/emi
|
||||
Set `SCIMESH_AUTO_START=0` to install without starting anything. The old demo
|
||||
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`):
|
||||
|
||||
|
||||
@@ -223,8 +223,16 @@ func runWithConfig(cfg infra.Config) error {
|
||||
|
||||
// deps.ready backs /health: readiness means the database answers, not just
|
||||
// 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)
|
||||
err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken))
|
||||
api := httptransport.NewServerWithOptions(
|
||||
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
|
||||
// LIFO, so the deferred stop() would fire *after* the wait below).
|
||||
|
||||
@@ -6,10 +6,17 @@ import (
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/agent"
|
||||
|
||||
"crypto/rand"
|
||||
"crypto/rsa"
|
||||
"crypto/sha256"
|
||||
"crypto/x509"
|
||||
"crypto/x509/pkix"
|
||||
"encoding/hex"
|
||||
"encoding/pem"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math/big"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
@@ -42,6 +49,10 @@ func runServe(args []string) error {
|
||||
email = flags.String("admin-email", "admin@scimesh.local", "admin account email")
|
||||
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>)")
|
||||
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 {
|
||||
return err
|
||||
@@ -96,27 +107,50 @@ func runServe(args []string) error {
|
||||
defer func() { _ = closeUsers() }()
|
||||
|
||||
// 5. Local worker agents before the server, so they can claim immediately.
|
||||
coordinatorURL := "http://" + *addr
|
||||
agents, err := spawnAgents(ctx, log, *dataDir, *workers, coordinatorURL, workerToken, venvPython)
|
||||
// They always dial the loopback address: an --addr of 0.0.0.0 is not a
|
||||
// connectable target from the same host.
|
||||
agentURL, resolvedPublic := serveURLs(*addr, *publicURL)
|
||||
agents, err := spawnAgents(ctx, log, *dataDir, *workers, agentURL, workerToken, venvPython)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stopAgents(agents)
|
||||
|
||||
// 6. The coordinator server itself.
|
||||
coordinatorPublicURL := *publicURL
|
||||
if coordinatorPublicURL == "" {
|
||||
coordinatorPublicURL = "http://" + *addr
|
||||
// TLS: explicit cert/key win; --tls-autogen creates a self-signed pair in
|
||||
// the data dir on first use (fingerprint printed for pinning).
|
||||
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{
|
||||
Addr: *addr,
|
||||
DatabaseEngine: "sqlite",
|
||||
DBPath: filepath.Join(*dataDir, "scimesh.db"),
|
||||
TLSCertFile: tlsCertFile,
|
||||
TLSKeyFile: tlsKeyFile,
|
||||
DisableRegistration: *noReg || os.Getenv("SCIMESH_DISABLE_REGISTRATION") == "1",
|
||||
Token: workerToken,
|
||||
JWTSecret: jwtSecret,
|
||||
UserserviceURL: "http://" + usersAddr,
|
||||
PublicCoordinatorURL: coordinatorPublicURL,
|
||||
PublicUserserviceURL: "http://" + usersAddr,
|
||||
PublicCoordinatorURL: resolvedPublic,
|
||||
// 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",
|
||||
StorageDir: filepath.Join(*dataDir, "artifacts"),
|
||||
DocsDir: *docsDir,
|
||||
@@ -132,12 +166,13 @@ func runServe(args []string) error {
|
||||
WorkerOfflineAfter: 1 * time.Minute,
|
||||
AutoMigrate: true,
|
||||
}
|
||||
browserURL, _ := serveURLs(*addr, *publicURL)
|
||||
if *open {
|
||||
openBrowser("http://" + *addr + "/ui/admin")
|
||||
openBrowser(browserURL + "/ui/admin")
|
||||
}
|
||||
|
||||
// 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)
|
||||
if runtimeStatus(venvPython) {
|
||||
fmt.Printf(" scientific runtime: ready (%s)\n", venvPython)
|
||||
@@ -337,3 +372,102 @@ func openBrowser(target string) {
|
||||
// #nosec G204 -- target is the local UI URL the operator asked to open.
|
||||
_ = 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
|
||||
}
|
||||
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)
|
||||
if err != nil {
|
||||
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 {
|
||||
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 {
|
||||
detail := err.Error()
|
||||
if strings.Contains(detail, "connection refused") {
|
||||
@@ -148,7 +148,7 @@ func CheckAuth(ctx context.Context, url, token, workerKey, userserviceURL string
|
||||
item.Detail = "no credential configured — will be checked at registration"
|
||||
return item
|
||||
}
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
client := &http.Client{Timeout: 30 * time.Second, Transport: tlsTransport(nil)}
|
||||
if workerKey != "" && userserviceURL != "" {
|
||||
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)))
|
||||
|
||||
@@ -48,16 +48,24 @@ type Client struct {
|
||||
}
|
||||
|
||||
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{
|
||||
baseURL: strings.TrimRight(baseURL, "/"),
|
||||
tokens: tokens,
|
||||
timeout: timeout,
|
||||
apiClient: &http.Client{
|
||||
Timeout: timeout,
|
||||
Transport: tlsTransport(nil),
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
},
|
||||
dlClient: &http.Client{
|
||||
Timeout: timeout,
|
||||
Timeout: transferTimeout,
|
||||
Transport: tlsTransport(nil),
|
||||
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
||||
if len(via) >= 10 {
|
||||
return fmt.Errorf("too many redirects")
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
@@ -223,3 +225,72 @@ func sha256Of(t *testing.T, value string) string {
|
||||
digest := sha256.Sum256([]byte(value))
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
// disables the docs route; the UI shows a hint page instead.
|
||||
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.
|
||||
MaxUploadBytes int64
|
||||
|
||||
@@ -123,6 +131,9 @@ func LoadConfig() (Config, error) {
|
||||
LogFile: os.Getenv("LOG_FILE"),
|
||||
StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"),
|
||||
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
|
||||
DBMaxConns: 10,
|
||||
DBConnectTimeout: 30 * time.Second,
|
||||
|
||||
@@ -13,17 +13,25 @@ import (
|
||||
const shutdownGrace = 15 * time.Second
|
||||
|
||||
// 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{
|
||||
Addr: addr,
|
||||
Handler: handler,
|
||||
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
|
||||
// (the ctx.Done branch below) — an unbuffered send would leak it forever.
|
||||
errCh := make(chan error, 1)
|
||||
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)
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
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"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
@@ -37,9 +38,29 @@ func Open(path string) (*sql.DB, error) {
|
||||
_ = db.Close()
|
||||
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
|
||||
}
|
||||
|
||||
// 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
|
||||
// method run identically inside or outside a transaction.
|
||||
type querier interface {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -347,3 +348,19 @@ func TestCancelByJobInvalidatesTasks(t *testing.T) {
|
||||
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
|
||||
// keeps the static basic-auth UI.
|
||||
userserviceURL string
|
||||
// disableRegistration forbids new accounts; login keeps working.
|
||||
disableRegistration bool
|
||||
// publicCoordinatorURL / publicUserserviceURL are the browser-facing URLs
|
||||
// rendered into the worker-enrollment command. Either may be empty; the
|
||||
// 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,
|
||||
maxUploadBytes int64, jwtSecret, userserviceURL string, m *metrics.Metrics, ready func(context.Context) error,
|
||||
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 {
|
||||
m = metrics.New()
|
||||
}
|
||||
@@ -88,6 +103,7 @@ func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval
|
||||
}
|
||||
return &Server{
|
||||
uc: uc,
|
||||
disableRegistration: opts.DisableRegistration,
|
||||
log: log,
|
||||
requestTimeout: requestTimeout,
|
||||
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
|
||||
// is embedded (serve mode): the key itself is the credential.
|
||||
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] != ""
|
||||
@@ -163,7 +179,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
|
||||
if s.uiSessionMode() {
|
||||
// Public auth pages — reachable without a session so a user can log in.
|
||||
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/register", s.handleUIRegisterForm)
|
||||
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
|
||||
// promotes or verifies it.
|
||||
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")
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
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"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
@@ -290,6 +291,17 @@ func indexOf(haystack, needle string) int {
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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)
|
||||
@@ -300,5 +312,9 @@ func Open(path string) (*sql.DB, error) {
|
||||
_ = db.Close()
|
||||
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
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"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,
|
||||
// 409 if the email is taken, or 400 on a malformed body / weak password.
|
||||
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
|
||||
if !decodeJSON(w, r, &req) {
|
||||
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.
|
||||
mux.HandleFunc("GET /health", h.handleHealth)
|
||||
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.
|
||||
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
|
||||
// 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.
|
||||
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("GET /worker-keys", chain(http.HandlerFunc(h.handleListWorkerKeys), 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)
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
$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"
|
||||
$Component = if ($env:SCIMESH_COMPONENT) { $env:SCIMESH_COMPONENT } else { "coordinator" }
|
||||
$Version = if ($env:SCIMESH_VERSION) { $env:SCIMESH_VERSION } else { "latest" }
|
||||
@@ -57,6 +62,63 @@ $Target = Join-Path $InstallDir "$Binary.exe"
|
||||
|
||||
Write-Host "Downloading $Url"
|
||||
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
|
||||
|
||||
Write-Host ""
|
||||
|
||||
+51
@@ -14,6 +14,13 @@
|
||||
set -eu
|
||||
|
||||
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}"
|
||||
VERSION="${SCIMESH_VERSION:-latest}"
|
||||
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}"
|
||||
echo "Downloading $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"
|
||||
mv "$TARGET.tmp" "$TARGET"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user