Report queued tasks the scheduling pass never considers

A task in retry backoff was filtered out before the candidate loop, so it
recorded no rejection at all: queued, apparently assignable, and silent. That is
the exact shape that made F5 take a live session to diagnose. It now reports
"retry backoff until <time>".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-26 23:42:26 +04:00
parent 0ead6d2d02
commit 4fbf3ac966
18 changed files with 1163 additions and 165 deletions
+165
View File
@@ -0,0 +1,165 @@
package authn
import (
"encoding/json"
"errors"
"net/http"
"orchestra/internal/authz"
"time"
)
const maxLoginBody = 8 << 10
type HTTP struct {
Users *Store
Sessions *authz.Sessions
SecureCookie bool
}
func (h HTTP) cookie(value string, maxAge int) *http.Cookie {
cookie := &http.Cookie{
Name: authz.SessionCookie,
Value: value,
Path: "/",
MaxAge: maxAge,
HttpOnly: true,
SameSite: http.SameSiteStrictMode,
Secure: h.SecureCookie,
}
if maxAge < 0 {
cookie.Expires = time.Unix(1, 0)
}
return cookie
}
func decode(w http.ResponseWriter, r *http.Request, dst any) error {
r.Body = http.MaxBytesReader(w, r.Body, maxLoginBody)
decoder := json.NewDecoder(r.Body)
decoder.DisallowUnknownFields()
if err := decoder.Decode(dst); err != nil {
return err
}
var extra any
if err := decoder.Decode(&extra); err == nil {
return errors.New("multiple JSON values")
}
return nil
}
func writeJSON(w http.ResponseWriter, status int, value any) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-store")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(value)
}
func (h HTTP) sessionUser(r *http.Request) (User, bool) {
if h.Users == nil || h.Sessions == nil {
return User{}, false
}
cookie, err := r.Cookie(authz.SessionCookie)
if err != nil {
return User{}, false
}
username, ok := h.Sessions.Username(cookie.Value)
if !ok {
return User{}, false
}
user, err := h.Users.User(username)
return user, err == nil
}
func (h HTTP) Session(w http.ResponseWriter, r *http.Request) {
if h.Users == nil || h.Sessions == nil {
http.Error(w, "login unavailable", http.StatusServiceUnavailable)
return
}
switch r.Method {
case http.MethodGet:
user, ok := h.sessionUser(r)
if !ok {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
writeJSON(w, http.StatusOK, user)
case http.MethodPost:
var body struct {
Username string `json:"username"`
Password string `json:"password"`
}
if err := decode(w, r, &body); err != nil {
http.Error(w, "invalid credentials", http.StatusUnauthorized)
return
}
user, err := h.Users.Authenticate(body.Username, body.Password)
if err != nil {
// Credential failures are deliberately indistinguishable. Database
// failures are not exposed either, but they remain a server error.
if errors.Is(err, ErrInvalidCredentials) {
http.Error(w, "invalid credentials", http.StatusUnauthorized)
} else {
http.Error(w, "login unavailable", http.StatusInternalServerError)
}
return
}
value, err := h.Sessions.IssueFor(user.Username)
if err != nil {
http.Error(w, "session unavailable", http.StatusInternalServerError)
return
}
http.SetCookie(w, h.cookie(value, int(h.Sessions.Duration().Seconds())))
writeJSON(w, http.StatusOK, user)
case http.MethodDelete:
if cookie, err := r.Cookie(authz.SessionCookie); err == nil {
h.Sessions.Revoke(cookie.Value)
}
http.SetCookie(w, h.cookie("", -1))
w.Header().Set("Cache-Control", "no-store")
w.WriteHeader(http.StatusNoContent)
default:
w.Header().Set("Allow", "GET, POST, DELETE")
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
func (h HTTP) Account(w http.ResponseWriter, r *http.Request) {
user, ok := h.sessionUser(r)
if !ok {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
switch r.Method {
case http.MethodGet:
writeJSON(w, http.StatusOK, user)
case http.MethodPut:
var body struct {
CurrentPassword string `json:"current_password"`
Username string `json:"username"`
NewPassword string `json:"new_password"`
}
if err := decode(w, r, &body); err != nil || body.CurrentPassword == "" {
http.Error(w, "current password is required", http.StatusBadRequest)
return
}
updated, err := h.Users.Update(user.Username, body.CurrentPassword, body.Username, body.NewPassword)
if err != nil {
switch {
case errors.Is(err, ErrInvalidCredentials):
http.Error(w, "current password is incorrect", http.StatusUnauthorized)
case errors.Is(err, ErrUsernameExists):
http.Error(w, err.Error(), http.StatusConflict)
default:
http.Error(w, err.Error(), http.StatusBadRequest)
}
return
}
// Credential changes revoke every browser holding this identity. The
// response carries the updated display name, then the UI signs in again.
h.Sessions.RevokeUser(user.Username)
http.SetCookie(w, h.cookie("", -1))
writeJSON(w, http.StatusOK, updated)
default:
w.Header().Set("Allow", "GET, PUT")
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
+94
View File
@@ -0,0 +1,94 @@
package authn
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"orchestra/internal/authz"
"testing"
)
func TestHTTPSessionLoginLookupAndLogout(t *testing.T) {
users, _ := openTestStore(t)
if _, _, err := users.SetPassword("kami", "correct horse battery"); err != nil {
t.Fatal(err)
}
h := HTTP{Users: users, Sessions: &authz.Sessions{}}
login := httptest.NewRequest(http.MethodPost, authz.SessionPath, bytes.NewBufferString(`{"username":"kami","password":"correct horse battery"}`))
w := httptest.NewRecorder()
h.Session(w, login)
if w.Code != http.StatusOK {
t.Fatalf("login status=%d body=%s", w.Code, w.Body)
}
response := w.Result()
cookies := response.Cookies()
if len(cookies) != 1 || cookies[0].Name != authz.SessionCookie || !cookies[0].HttpOnly {
t.Fatalf("cookies=%+v", cookies)
}
var account User
if err := json.Unmarshal(w.Body.Bytes(), &account); err != nil || account.Username != "kami" {
t.Fatalf("account=%+v err=%v", account, err)
}
lookup := httptest.NewRequest(http.MethodGet, authz.SessionPath, nil)
lookup.AddCookie(cookies[0])
w = httptest.NewRecorder()
h.Session(w, lookup)
if w.Code != http.StatusOK || !bytes.Contains(w.Body.Bytes(), []byte(`"username":"kami"`)) {
t.Fatalf("lookup status=%d body=%s", w.Code, w.Body)
}
logout := httptest.NewRequest(http.MethodDelete, authz.SessionPath, nil)
logout.AddCookie(cookies[0])
w = httptest.NewRecorder()
h.Session(w, logout)
if w.Code != http.StatusNoContent || h.Sessions.Valid(cookies[0].Value) {
t.Fatalf("logout status=%d valid=%v", w.Code, h.Sessions.Valid(cookies[0].Value))
}
}
func TestHTTPLoginDoesNotRevealUnknownUsername(t *testing.T) {
users, _ := openTestStore(t)
if _, _, err := users.SetPassword("kami", "correct horse battery"); err != nil {
t.Fatal(err)
}
h := HTTP{Users: users, Sessions: &authz.Sessions{}}
for _, body := range []string{
`{"username":"kami","password":"wrong password"}`,
`{"username":"unknown","password":"wrong password"}`,
} {
w := httptest.NewRecorder()
h.Session(w, httptest.NewRequest(http.MethodPost, authz.SessionPath, bytes.NewBufferString(body)))
if w.Code != http.StatusUnauthorized || w.Body.String() != "invalid credentials\n" {
t.Fatalf("status=%d body=%q", w.Code, w.Body.String())
}
}
}
func TestHTTPAccountUpdateRevokesExistingSessions(t *testing.T) {
users, _ := openTestStore(t)
if _, _, err := users.SetPassword("operator", "original password"); err != nil {
t.Fatal(err)
}
sessions := &authz.Sessions{}
h := HTTP{Users: users, Sessions: sessions}
value, err := sessions.IssueFor("operator")
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPut, "/v1/ui/account", bytes.NewBufferString(`{"current_password":"original password","username":"kami","new_password":"replacement password"}`))
req.AddCookie(&http.Cookie{Name: authz.SessionCookie, Value: value})
w := httptest.NewRecorder()
h.Account(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", w.Code, w.Body)
}
if sessions.Valid(value) {
t.Fatal("credential update retained an old browser session")
}
if _, err := users.Authenticate("kami", "replacement password"); err != nil {
t.Fatalf("updated login: %v", err)
}
}
+343
View File
@@ -0,0 +1,343 @@
// Package authn owns browser-operator identities and credential verification.
// Authorization policy remains in authz; this package only proves who signed
// in. Operator records live in a small embedded bbolt database so a deployment
// never needs to carry a reusable password hash in its environment.
package authn
import (
"crypto/subtle"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"time"
"unicode"
"unicode/utf8"
bolt "go.etcd.io/bbolt"
"golang.org/x/crypto/bcrypt"
)
const (
DatabaseFile = "auth.db"
MinimumPassword = 10
maximumPassword = 72 // bcrypt rejects passwords longer than 72 bytes.
maximumUsername = 128
databaseOpenWait = 2 * time.Second
databaseFileMode = 0600
databaseDirectory = 0700
)
var (
usersBucket = []byte("operator_users")
ErrInvalidCredentials = errors.New("invalid username or password")
ErrUsernameExists = errors.New("username already exists")
// Unknown users still take a bcrypt comparison. The hash is generated once
// at process start with the same cost used for real records so the login
// response does not disclose whether an account exists.
dummyPasswordHash = func() []byte {
hash, err := bcrypt.GenerateFromPassword([]byte("orchestra-invalid-login-sentinel"), bcrypt.DefaultCost)
if err != nil {
panic(err)
}
return hash
}()
)
type User struct {
Username string `json:"username"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type storedUser struct {
User
PasswordHash string `json:"password_hash"`
}
type Store struct {
db *bolt.DB
}
func Path(dataDir string) string { return filepath.Join(dataDir, DatabaseFile) }
func Open(path string) (*Store, error) {
if strings.TrimSpace(path) == "" {
return nil, errors.New("auth database path is required")
}
if err := os.MkdirAll(filepath.Dir(path), databaseDirectory); err != nil {
return nil, fmt.Errorf("create auth database directory: %w", err)
}
db, err := bolt.Open(path, databaseFileMode, &bolt.Options{Timeout: databaseOpenWait})
if err != nil {
return nil, fmt.Errorf("open auth database: %w", err)
}
s := &Store{db: db}
if err := db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(usersBucket)
return err
}); err != nil {
_ = db.Close()
return nil, fmt.Errorf("initialize auth database: %w", err)
}
return s, nil
}
func (s *Store) Close() error {
if s == nil || s.db == nil {
return nil
}
return s.db.Close()
}
func normalizedUsername(username string) string {
return strings.ToLower(strings.TrimSpace(username))
}
func ValidateUsername(username string) error {
username = strings.TrimSpace(username)
if username == "" {
return errors.New("username is required")
}
if len(username) > maximumUsername {
return fmt.Errorf("username must be at most %d bytes", maximumUsername)
}
if !utf8.ValidString(username) {
return errors.New("username must be valid UTF-8")
}
for _, r := range username {
if unicode.IsControl(r) {
return errors.New("username must not contain control characters")
}
}
return nil
}
func ValidatePassword(password string) error {
if len(password) < MinimumPassword {
return fmt.Errorf("password must be at least %d characters", MinimumPassword)
}
if len([]byte(password)) > maximumPassword {
return fmt.Errorf("password must be at most %d bytes", maximumPassword)
}
return nil
}
func decodeUser(raw []byte) (storedUser, error) {
var user storedUser
if err := json.Unmarshal(raw, &user); err != nil {
return storedUser{}, err
}
return user, nil
}
func (s *Store) Count() (int, error) {
count := 0
err := s.db.View(func(tx *bolt.Tx) error {
count = tx.Bucket(usersBucket).Stats().KeyN
return nil
})
return count, err
}
func (s *Store) Users() ([]User, error) {
users := []User{}
err := s.db.View(func(tx *bolt.Tx) error {
return tx.Bucket(usersBucket).ForEach(func(_, raw []byte) error {
stored, err := decodeUser(raw)
if err != nil {
return err
}
users = append(users, stored.User)
return nil
})
})
return users, err
}
func (s *Store) User(username string) (User, error) {
var out User
err := s.db.View(func(tx *bolt.Tx) error {
raw := tx.Bucket(usersBucket).Get([]byte(normalizedUsername(username)))
if raw == nil {
return ErrInvalidCredentials
}
stored, err := decodeUser(raw)
if err == nil {
out = stored.User
}
return err
})
return out, err
}
// SetPassword creates an operator or replaces that operator's password. It is
// intended for the local orchestra-user command; browser changes use Update,
// which also proves the current password.
func (s *Store) SetPassword(username, password string) (User, bool, error) {
username = strings.TrimSpace(username)
if err := ValidateUsername(username); err != nil {
return User{}, false, err
}
if err := ValidatePassword(password); err != nil {
return User{}, false, err
}
hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
if err != nil {
return User{}, false, err
}
return s.putHash(username, string(hash), false)
}
// ImportBcrypt is the one-time compatibility bridge from the old environment
// credential. It only creates the named user when the database is empty.
func (s *Store) ImportBcrypt(username, passwordHash string) (bool, error) {
username = strings.TrimSpace(username)
if err := ValidateUsername(username); err != nil {
return false, err
}
if _, err := bcrypt.Cost([]byte(passwordHash)); err != nil {
return false, fmt.Errorf("legacy web password hash must be bcrypt: %w", err)
}
_, created, err := s.putHash(username, passwordHash, true)
return created, err
}
func (s *Store) putHash(username, passwordHash string, onlyIfEmpty bool) (User, bool, error) {
now := time.Now().UTC()
key := []byte(normalizedUsername(username))
var out User
created := false
err := s.db.Update(func(tx *bolt.Tx) error {
bucket := tx.Bucket(usersBucket)
if onlyIfEmpty && bucket.Stats().KeyN != 0 {
return nil
}
stored := storedUser{User: User{Username: username, CreatedAt: now, UpdatedAt: now}, PasswordHash: passwordHash}
if raw := bucket.Get(key); raw != nil {
current, err := decodeUser(raw)
if err != nil {
return err
}
stored.CreatedAt = current.CreatedAt
} else {
created = true
}
encoded, err := json.Marshal(stored)
if err != nil {
return err
}
out = stored.User
return bucket.Put(key, encoded)
})
return out, created, err
}
func (s *Store) Authenticate(username, password string) (User, error) {
var stored storedUser
found := false
err := s.db.View(func(tx *bolt.Tx) error {
raw := tx.Bucket(usersBucket).Get([]byte(normalizedUsername(username)))
if raw == nil {
return nil
}
var err error
stored, err = decodeUser(raw)
found = err == nil
return err
})
if err != nil {
return User{}, err
}
hash := dummyPasswordHash
if found {
hash = []byte(stored.PasswordHash)
}
passwordOK := bcrypt.CompareHashAndPassword(hash, []byte(password)) == nil
usernameOK := found && subtle.ConstantTimeCompare(
[]byte(normalizedUsername(username)),
[]byte(normalizedUsername(stored.Username)),
) == 1
if !passwordOK || !usernameOK {
return User{}, ErrInvalidCredentials
}
return stored.User, nil
}
// Update changes the authenticated operator's username and/or password. The
// current password is required even though the endpoint also requires a live
// browser session, protecting an unattended unlocked browser.
func (s *Store) Update(currentUsername, currentPassword, newUsername, newPassword string) (User, error) {
current, err := s.Authenticate(currentUsername, currentPassword)
if err != nil {
return User{}, err
}
newUsername = strings.TrimSpace(newUsername)
if newUsername == "" {
newUsername = current.Username
}
if err := ValidateUsername(newUsername); err != nil {
return User{}, err
}
var newHash string
if newPassword != "" {
if err := ValidatePassword(newPassword); err != nil {
return User{}, err
}
hash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost)
if err != nil {
return User{}, err
}
newHash = string(hash)
}
oldKey := []byte(normalizedUsername(current.Username))
newKey := []byte(normalizedUsername(newUsername))
var out User
err = s.db.Update(func(tx *bolt.Tx) error {
bucket := tx.Bucket(usersBucket)
raw := bucket.Get(oldKey)
if raw == nil {
return ErrInvalidCredentials
}
stored, err := decodeUser(raw)
if err != nil {
return err
}
// Refuse a stale credential update if another password change landed
// between Authenticate and this write transaction.
if bcrypt.CompareHashAndPassword([]byte(stored.PasswordHash), []byte(currentPassword)) != nil {
return ErrInvalidCredentials
}
if !bytesEqual(oldKey, newKey) && bucket.Get(newKey) != nil {
return ErrUsernameExists
}
stored.Username = newUsername
stored.UpdatedAt = time.Now().UTC()
if newHash != "" {
stored.PasswordHash = newHash
}
encoded, err := json.Marshal(stored)
if err != nil {
return err
}
if !bytesEqual(oldKey, newKey) {
if err := bucket.Delete(oldKey); err != nil {
return err
}
}
if err := bucket.Put(newKey, encoded); err != nil {
return err
}
out = stored.User
return nil
})
return out, err
}
func bytesEqual(a, b []byte) bool {
return len(a) == len(b) && subtle.ConstantTimeCompare(a, b) == 1
}
+108
View File
@@ -0,0 +1,108 @@
package authn
import (
"errors"
"path/filepath"
"testing"
"golang.org/x/crypto/bcrypt"
)
func openTestStore(t *testing.T) (*Store, string) {
t.Helper()
path := filepath.Join(t.TempDir(), DatabaseFile)
store, err := Open(path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
return store, path
}
func TestPasswordRecordPersistsAndAuthenticates(t *testing.T) {
store, path := openTestStore(t)
created, wasCreated, err := store.SetPassword("Kami", "correct horse battery")
if err != nil || !wasCreated || created.Username != "Kami" {
t.Fatalf("created=%+v new=%v err=%v", created, wasCreated, err)
}
if _, err := store.Authenticate("KAMI", "correct horse battery"); err != nil {
t.Fatalf("authenticate: %v", err)
}
if _, err := store.Authenticate("Kami", "wrong password"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("wrong password error = %v", err)
}
if err := store.Close(); err != nil {
t.Fatal(err)
}
reopened, err := Open(path)
if err != nil {
t.Fatal(err)
}
defer reopened.Close()
if _, err := reopened.Authenticate("kami", "correct horse battery"); err != nil {
t.Fatalf("persisted authentication: %v", err)
}
}
func TestUpdateRequiresCurrentPasswordAndMovesUsername(t *testing.T) {
store, _ := openTestStore(t)
if _, _, err := store.SetPassword("operator", "original password"); err != nil {
t.Fatal(err)
}
if _, err := store.Update("operator", "wrong password", "kami", "replacement password"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("wrong current password error = %v", err)
}
updated, err := store.Update("operator", "original password", "kami", "replacement password")
if err != nil || updated.Username != "kami" {
t.Fatalf("updated=%+v err=%v", updated, err)
}
if _, err := store.Authenticate("operator", "original password"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("old credential error = %v", err)
}
if _, err := store.Authenticate("kami", "replacement password"); err != nil {
t.Fatalf("new credential: %v", err)
}
}
func TestUpdateRefusesExistingUsername(t *testing.T) {
store, _ := openTestStore(t)
if _, _, err := store.SetPassword("one", "password one"); err != nil {
t.Fatal(err)
}
if _, _, err := store.SetPassword("two", "password two"); err != nil {
t.Fatal(err)
}
if _, err := store.Update("one", "password one", "TWO", ""); !errors.Is(err, ErrUsernameExists) {
t.Fatalf("collision error = %v", err)
}
}
func TestLegacyHashImportsOnlyIntoEmptyDatabase(t *testing.T) {
store, _ := openTestStore(t)
hash, err := bcrypt.GenerateFromPassword([]byte("legacy password"), bcrypt.MinCost)
if err != nil {
t.Fatal(err)
}
if imported, err := store.ImportBcrypt("legacy", string(hash)); err != nil || !imported {
t.Fatalf("imported=%v err=%v", imported, err)
}
if imported, err := store.ImportBcrypt("intruder", string(hash)); err != nil || imported {
t.Fatalf("second import=%v err=%v", imported, err)
}
if _, err := store.Authenticate("legacy", "legacy password"); err != nil {
t.Fatalf("imported credential: %v", err)
}
}
func TestCredentialValidation(t *testing.T) {
store, _ := openTestStore(t)
if _, _, err := store.SetPassword("", "a sufficiently long password"); err == nil {
t.Fatal("blank username accepted")
}
if _, _, err := store.SetPassword("operator", "short"); err == nil {
t.Fatal("short password accepted")
}
if _, _, err := store.SetPassword("operator", string(make([]byte, maximumPassword+1))); err == nil {
t.Fatal("oversized bcrypt password accepted")
}
}