package api import ( "bytes" "context" "encoding/json" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "sync" "testing" "time" "drivervault/apiserver/internal/config" "drivervault/apiserver/internal/ocpp" "drivervault/apiserver/internal/pb" ) // This test exercises the two PocketBase-touching control paths end-to-end // through the real Handler + middleware chain: step-up re-authentication for a // destructive action (reset), and best-effort audit persistence to the // control_audit collection. A stand-in PocketBase (fakePB) serves exactly the // endpoints the flow hits; a simulated charge point connects to the live CSMS so // the command has a real session to reach. const ( e2eUserID = "u1" e2eUserMail = "owner@test.local" e2ePassword = "correct-horse-battery" e2eSerial = "CP-TEST" e2eToken = "charger-control-token-abc" e2eBearer = "user-bearer-token" ) // fakePB records audit creates and answers the identity, user-record, // re-auth and service-auth calls the control flow makes. type fakePB struct { mu sync.Mutex audits []map[string]any userPluginSettings json.RawMessage } func (f *fakePB) handler(t *testing.T) http.Handler { mux := http.NewServeMux() // Service-account auth (pb.Client.Authenticate). mux.HandleFunc("POST /api/collections/_superusers/auth-with-password", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"token": "svc-token"}) }) // Identify the bearer (withAuth → identify → AuthRefresh). mux.HandleFunc("POST /api/collections/users/auth-refresh", func(w http.ResponseWriter, r *http.Request) { if r.Header.Get("Authorization") == "" { writeJSON(w, 401, map[string]any{}) return } writeJSON(w, 200, map[string]any{"record": map[string]any{ "id": e2eUserID, "email": e2eUserMail, "name": "Owner", "role": "user", "organization": "", }}) }) // User record (userPluginSettings + callerForUser both GET this). mux.HandleFunc("GET /api/collections/users/records/"+e2eUserID, func(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{ "id": e2eUserID, "email": e2eUserMail, "role": "user", "organization": "", "pluginSettings": f.userPluginSettings, }) }) // Step-up re-auth (reauthenticate → LoginWithPassword). mux.HandleFunc("POST /api/collections/users/auth-with-password", func(w http.ResponseWriter, r *http.Request) { var body struct{ Identity, Password string } _ = json.NewDecoder(r.Body).Decode(&body) if body.Identity == e2eUserMail && body.Password == e2ePassword { writeJSON(w, 200, map[string]any{"token": "reauth-ok", "record": map[string]any{"id": e2eUserID}}) return } writeJSON(w, 400, map[string]any{"message": "Failed to authenticate."}) }) // Durable audit sink. mux.HandleFunc("POST /api/collections/control_audit/records", func(w http.ResponseWriter, r *http.Request) { var rec map[string]any _ = json.NewDecoder(r.Body).Decode(&rec) f.mu.Lock() f.audits = append(f.audits, rec) f.mu.Unlock() writeJSON(w, 200, map[string]any{"id": "audit-rec"}) }) return mux } func (f *fakePB) auditsWith(action, result string) int { f.mu.Lock() defer f.mu.Unlock() n := 0 for _, a := range f.audits { if a["action"] == action && strings.Contains(toStr(a["result"]), result) { n++ } } return n } func toStr(v any) string { s, _ := v.(string) return s } func TestControlStepUpAndAuditE2E(t *testing.T) { tokenHashHex := hashToken(e2eToken) fake := &fakePB{ userPluginSettings: json.RawMessage(`{"ankerSolix":{"enabled":true,"config":{"controlMode":"own"},` + `"controlChargers":{"` + e2eSerial + `":{"tokenHash":"` + tokenHashHex + `","tokenHint":"` + tokenHint(e2eToken) + `"}}}}`), } pbSrv := httptest.NewServer(fake.handler(t)) defer pbSrv.Close() // Enable the anker-solix plugin globally via a plugins.json. dir := t.TempDir() pluginsFile := filepath.Join(dir, "plugins.json") if err := os.WriteFile(pluginsFile, []byte(`{"anker-solix":{"enabled":true}}`), 0o600); err != nil { t.Fatal(err) } s := New(config.Config{ UsersCollection: "users", PluginsFile: pluginsFile, OCPPRequireTLS: false, // httptest is plaintext; TLS enforcement covered elsewhere }, pb.New(pbSrv.URL, "admin@test.local", "pw")) if err := s.plugins.Load(); err != nil { t.Fatalf("load plugins: %v", err) } // Seed the control-token index (mirrors what generating a token does). s.control.setUser(e2eUserID, map[string]ankerChargerBinding{ e2eSerial: {TokenHash: tokenHashHex, TokenHint: tokenHint(e2eToken)}, }) appSrv := httptest.NewServer(s.Handler()) defer appSrv.Close() // Connect a simulated charger to the live CSMS (own mode). stop := connectCharger(t, "ws"+strings.TrimPrefix(appSrv.URL, "http")+"/ocpp/"+e2eSerial, e2eSerial, e2eToken) defer stop() // Wait for the session to register. if !waitFor(2*time.Second, func() bool { _, ok := s.ocpp.SessionFor(e2eSerial); return ok }) { t.Fatal("charger session never registered on the CSMS") } // 1. Destructive reset without confirm → 400. if code, _ := postControl(t, appSrv.URL, "reset", map[string]any{}); code != http.StatusBadRequest { t.Errorf("reset without confirm: status %d, want 400", code) } // 2. Confirmed reset with the WRONG password → 401 (step-up fails). if code, _ := postControl(t, appSrv.URL, "reset", map[string]any{"confirm": true, "password": "wrong"}); code != http.StatusUnauthorized { t.Errorf("reset with wrong password: status %d, want 401", code) } // 3. Confirmed reset with the CORRECT password → 200 and the charger accepts. code, body := postControl(t, appSrv.URL, "reset", map[string]any{"confirm": true, "password": e2ePassword}) if code != http.StatusOK { t.Fatalf("reset with correct password: status %d body %s, want 200", code, body) } if got := body["status"]; got != "Accepted" { t.Errorf("reset status = %v, want Accepted", got) } // The audit sink must have recorded the failed re-auth and the accepted reset // (persisted asynchronously, so poll). if !waitFor(2*time.Second, func() bool { return fake.auditsWith("reset", "reauth-failed") >= 1 }) { t.Error("expected a control_audit record for the failed re-auth") } if !waitFor(2*time.Second, func() bool { return fake.auditsWith("reset", "Accepted") >= 1 }) { t.Error("expected a control_audit record for the accepted reset") } } // postControl issues an authenticated control command and returns the status and // decoded JSON body. func postControl(t *testing.T, base, action string, payload map[string]any) (int, map[string]any) { t.Helper() b, _ := json.Marshal(payload) req, _ := http.NewRequest(http.MethodPost, base+"/api/integrations/anker-solix/chargers/"+e2eSerial+"/"+action, bytes.NewReader(b)) req.Header.Set("Authorization", e2eBearer) req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("%s: %v", action, err) } defer resp.Body.Close() var out map[string]any _ = json.NewDecoder(resp.Body).Decode(&out) return resp.StatusCode, out } // connectCharger dials the CSMS as an OCPP charge point and answers every inbound // control CALL with {status:"Accepted"}. Returns a stop function. func connectCharger(t *testing.T, url, serial, token string) func() { t.Helper() h := http.Header{} h.Set("Authorization", ocpp.BasicAuthHeader(serial, token)) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() conn, err := ocpp.Dial(ctx, url, []string{"ocpp1.6"}, h) if err != nil { t.Fatalf("charger dial: %v", err) } go func() { for { data, err := conn.ReadMessage() if err != nil { return } msg, err := ocpp.DecodeMessage(data) if err != nil || msg.Type != ocpp.MessageTypeCall { continue } out, _ := ocpp.EncodeCallResult(msg.ID, map[string]any{"status": "Accepted"}) _ = conn.WriteMessage(out) } }() return func() { _ = conn.Close() } } func waitFor(d time.Duration, cond func() bool) bool { deadline := time.Now().Add(d) for time.Now().Before(deadline) { if cond() { return true } time.Sleep(10 * time.Millisecond) } return cond() }