package api import ( "context" "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "log" "net/http" "net/url" "strings" "sync" "time" "drivervault/apiserver/internal/ocpp" ) // This file is the control half of the Anker Solix integration: the OCPP // WebSocket endpoint a charger dials out to, the per-charger control-token index // that authenticates it, and the REST endpoints the web app calls to issue OCPP // commands. Monitoring (the read-only cloud plugin) lives in // integrations_ankersolix.go and the OCPP protocol itself in internal/ocpp. // ---- control-token index ----------------------------------------------------- // ankerControlBinding is what a per-charger control token resolves to. type ankerControlBinding struct { UserID string Serial string } // controlIndex maps a control token's SHA-256 hash to its owning user + charger. // Only hashes are held (in memory and at rest), so a leak of the index or the // stored settings never yields a usable charger credential. It is seeded when a // token is (re)generated and rebuilt from PocketBase the first time an unknown // token connects (or lazily at startup). type controlIndex struct { mu sync.RWMutex byHash map[string]ankerControlBinding built bool } func newControlIndex() *controlIndex { return &controlIndex{byHash: map[string]ankerControlBinding{}} } // lookup resolves the presented plaintext token by hashing it and matching a // stored hash — so the plaintext is never compared or retained. func (ci *controlIndex) lookup(token string) (ankerControlBinding, bool) { ci.mu.RLock() defer ci.mu.RUnlock() b, ok := ci.byHash[hashToken(token)] return b, ok } // setUser replaces all of a user's token entries with the given charger set, so // a regenerated or revoked token immediately stops resolving. func (ci *controlIndex) setUser(userID string, chargers map[string]ankerChargerBinding) { ci.mu.Lock() defer ci.mu.Unlock() for hash, b := range ci.byHash { if b.UserID == userID { delete(ci.byHash, hash) } } for serial, cb := range chargers { if cb.TokenHash != "" { ci.byHash[cb.TokenHash] = ankerControlBinding{UserID: userID, Serial: serial} } } } // ensureControlIndex builds the token index from every user's pluginSettings the // first time it is needed. Best effort: a failure leaves built=false so a later // connect retries. func (s *Server) ensureControlIndex(ctx context.Context) { s.control.mu.RLock() built := s.control.built s.control.mu.RUnlock() if built || !s.pb.Configured() { return } q := url.Values{} q.Set("perPage", "500") q.Set("fields", "id,pluginSettings") lr, err := s.pb.List(ctx, s.usersCollection(), q) if err != nil { return } var items []struct { ID string `json:"id"` PluginSettings json.RawMessage `json:"pluginSettings"` } _ = json.Unmarshal(lr.Items, &items) for _, it := range items { var doc ankerSettingsDoc if len(it.PluginSettings) > 0 { _ = json.Unmarshal(it.PluginSettings, &doc) } if len(doc.AnkerSolix.ControlChargers) > 0 { s.control.setUser(it.ID, doc.AnkerSolix.ControlChargers) } } s.control.mu.Lock() s.control.built = true s.control.mu.Unlock() } // ---- rate limiting ----------------------------------------------------------- // rateLimiter is a simple fixed-window limiter keyed by an arbitrary string // (here user id + charger serial). It bounds how fast control commands can be // issued so a valid session can't hammer Start/Stop/Reset at a physical charger. // // It is intentionally in-memory and resets on restart: the window is one minute, // so a restart clears at most a sub-minute budget, and persisting per-key // counters would add write amplification for negligible security value. type rateLimiter struct { mu sync.Mutex window time.Duration max int hits map[string]*rlEntry } type rlEntry struct { start time.Time count int } func newRateLimiter(max int, window time.Duration) *rateLimiter { return &rateLimiter{window: window, max: max, hits: map[string]*rlEntry{}} } func (rl *rateLimiter) allow(key string) bool { rl.mu.Lock() defer rl.mu.Unlock() now := time.Now() e := rl.hits[key] if e == nil || now.Sub(e.start) > rl.window { rl.hits[key] = &rlEntry{start: now, count: 1} return true } if e.count >= rl.max { return false } e.count++ return true } // ---- audit ------------------------------------------------------------------- // auditControl emits a single structured audit line for a control-plane event. // Control commands move a physical actuator, so every one records who did what // to which charger, with the outcome. (Durable persistence to PocketBase is a // planned follow-up; this greppable line is the first-cut trail.) func (s *Server) auditControl(who *callerIdentity, serial, action string, params map[string]any, result string, actErr error) { userID, orgID := "", "" if who != nil { userID, orgID = who.ID, who.OrgID } entry := map[string]any{ "audit": "anker-control", "ts": time.Now().UTC().Format(time.RFC3339), "serial": serial, "action": action, "result": result, "userId": userID, } if orgID != "" { entry["orgId"] = orgID } if len(params) > 0 { entry["params"] = params } if actErr != nil { entry["error"] = actErr.Error() } b, _ := json.Marshal(entry) log.Printf("AUDIT %s", b) // Durable, best-effort: persist to the control_audit collection if present. // The structured log line above is the fallback when it is not, so a failure // here never blocks or fails the control action. if !s.pb.Configured() { return } rec := map[string]any{ "user_id": userID, "org_id": orgID, "serial": serial, "action": action, "result": result, } auditParams := params if actErr != nil { auditParams = map[string]any{"error": actErr.Error()} for k, v := range params { auditParams[k] = v } } if len(auditParams) > 0 { rec["params"] = auditParams } go func() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := s.pb.Create(ctx, colControlAudit, rec, nil); err != nil { log.Printf("audit: could not persist control event (serial=%s action=%s): %v", serial, action, err) } }() } // ---- OCPP WebSocket endpoint ------------------------------------------------- // handleOCPPConnect is the WebSocket endpoint a charger dials out to. It // authenticates the charger with OCPP Basic auth (username = serial, password = // the per-charger control token), resolves the owner's control mode, and either // registers the session (own) or bridges it to Anker's cloud (proxy). func (s *Server) handleOCPPConnect(w http.ResponseWriter, r *http.Request) { serial := r.PathValue("serial") // A plaintext ws:// connection would carry the charger's Basic-auth token in // the clear, so reject it unless TLS was explicitly made optional (dev). s.mu.RLock() requireTLS := s.cfg.OCPPRequireTLS s.mu.RUnlock() if requireTLS && !requestIsSecure(r) { writeError(w, http.StatusForbidden, "OCPP connections must use TLS (wss)") return } user, token, ok := r.BasicAuth() if !ok || token == "" { w.Header().Set("WWW-Authenticate", `Basic realm="ocpp"`) writeError(w, http.StatusUnauthorized, "charger authentication required") return } binding, found := s.control.lookup(token) if !found { s.ensureControlIndex(r.Context()) binding, found = s.control.lookup(token) } if !found || binding.Serial != serial || (user != "" && user != serial) { writeError(w, http.StatusUnauthorized, "invalid charger credentials") return } who, userRaw, err := s.callerForUser(r.Context(), binding.UserID) if err != nil || who == nil { writeError(w, http.StatusForbidden, "charger owner unavailable") return } res := s.resolveAnker(r.Context(), who, userRaw) if !res.available || !res.orgEnabled || !res.enabled { writeError(w, http.StatusForbidden, "the charger owner has not enabled the integration") return } mode := res.eff.ControlMode if mode != ankerControlOwn && mode != ankerControlProxy { writeError(w, http.StatusForbidden, "control mode is off for this charger") return } // Resolve the proxy upstream before upgrading so a failure is a clean HTTP // error rather than a dropped WebSocket. var upstreamURL, upstreamAuth string if mode == ankerControlProxy { upstreamURL, upstreamAuth, err = s.ankerProxyUpstream(r.Context(), res, serial) if err != nil { writeError(w, http.StatusBadGateway, "cannot reach Anker upstream for proxy mode: "+err.Error()) return } } conn, err := ocpp.Upgrade(w, r) if err != nil { // The response may already be hijacked; just log and drop. log.Printf("ocpp: upgrade failed for %s: %v", serial, err) return } if mode == ankerControlOwn { if _, err := s.ocpp.Accept(serial, ocpp.ModeOwn, conn, nil); err != nil { log.Printf("ocpp: accept own %s: %v", serial, err) _ = conn.Close() return } s.auditControl(who, serial, "ocpp.connect.own", nil, "accepted", nil) return } up, err := ocpp.DialUpstream(r.Context(), upstreamURL, upstreamAuth) if err != nil { log.Printf("ocpp: proxy upstream dial failed for %s: %v", serial, err) _ = conn.Close() return } if _, err := s.ocpp.Accept(serial, ocpp.ModeProxy, conn, up); err != nil { log.Printf("ocpp: accept proxy %s: %v", serial, err) _ = conn.Close() _ = up.Close() return } s.auditControl(who, serial, "ocpp.connect.proxy", nil, "accepted", nil) } // callerForUser builds a callerIdentity (and returns the pluginSettings blob) for // a user id, so the OCPP endpoint can resolve the cascade without a bearer token. func (s *Server) callerForUser(ctx context.Context, userID string) (*callerIdentity, json.RawMessage, error) { if userID == "" || !s.pb.Configured() { return nil, nil, errors.New("not configured") } var rec struct { ID string `json:"id"` Role string `json:"role"` Organization string `json:"organization"` PluginSettings json.RawMessage `json:"pluginSettings"` } if err := s.pb.GetOne(ctx, s.usersCollection(), userID, &rec); err != nil { return nil, nil, err } role := rec.Role if role == "" { role = roleUser } return &callerIdentity{ID: rec.ID, Role: role, OrgID: rec.Organization}, rec.PluginSettings, nil } // ankerProxyUpstream asks the read-only cloud plugin for the charger's OCPP // endpoint (its "ocpp-info" capability) and extracts a ws(s):// URL from the // reply. Anker's own per-charger auth key is not available to us, so no // Authorization header is set — if the upstream requires one it will reject the // dial, surfaced as a clear error. func (s *Server) ankerProxyUpstream(ctx context.Context, res ankerResolution, serial string) (string, string, error) { if strings.TrimSpace(res.eff.Email) == "" || strings.TrimSpace(res.eff.Password) == "" { return "", "", errors.New("Anker account credentials are required for proxy mode") } cfg := map[string]string{"email": res.eff.Email, "password": res.eff.Password, "country": res.eff.Country} payload, _ := json.Marshal(map[string]string{"sn": serial}) raw, err := s.plugins.InvokeWith(ctx, ankerPlugin, cfg, "ocpp-info", payload) if err != nil { return "", "", err } u := extractWSURL(raw) if u == "" { return "", "", errors.New("no OCPP endpoint URL in the Anker response") } if !ankerUpstreamAllowed(u) { return "", "", errors.New("Anker returned an OCPP endpoint on an untrusted host") } return u, "", nil } // ---- control REST endpoints -------------------------------------------------- // handleAnkerControlStatus reports the control state for one charger: the // resolved mode, whether it is connected to our CSMS, the provisioning endpoint // and token, and a live status snapshot when connected. func (s *Server) handleAnkerControlStatus(w http.ResponseWriter, r *http.Request) { who := caller(r) if who == nil { writeError(w, http.StatusUnauthorized, "not authenticated") return } sn := r.PathValue("sn") userRaw := s.userPluginSettings(r.Context(), who.ID) res := s.resolveAnker(r.Context(), who, userRaw) binding := ankerBindingFor(userRaw, sn) body := map[string]any{ "serial": sn, "controlMode": res.eff.ControlMode, "available": res.available && res.orgEnabled && res.enabled, "endpoint": s.ocppEndpoint(r, sn), "hasToken": binding.TokenHash != "", "tokenHint": binding.TokenHint, // last-4 only; the token is shown once at generation } if sess, ok := s.ocpp.SessionFor(sn); ok { body["connected"] = true body["status"] = sess.Snapshot() } else { body["connected"] = false } writeJSON(w, http.StatusOK, body) } // handleAnkerControlToken (re)generates the per-charger control token the // operator installs into the charger, and updates the token index. func (s *Server) handleAnkerControlToken(w http.ResponseWriter, r *http.Request) { who := caller(r) if who == nil { writeError(w, http.StatusUnauthorized, "not authenticated") return } if !s.pb.Configured() { writeError(w, http.StatusServiceUnavailable, "integration settings not configured on the server") return } sn := r.PathValue("sn") if strings.TrimSpace(sn) == "" { writeError(w, http.StatusBadRequest, "missing charger serial") return } token := genControlToken() userRaw := s.userPluginSettings(r.Context(), who.ID) var updated map[string]ankerChargerBinding newDoc := mergeAnker(userRaw, func(as *ankerStored) { if as.ControlChargers == nil { as.ControlChargers = map[string]ankerChargerBinding{} } as.ControlChargers[sn] = ankerChargerBinding{ TokenHash: hashToken(token), TokenHint: tokenHint(token), AddedAt: time.Now().UTC().Format(time.RFC3339), } updated = as.ControlChargers }) if err := s.pb.Update(r.Context(), s.usersCollection(), who.ID, map[string]any{"pluginSettings": newDoc}, nil); err != nil { writePBError(w, err) return } s.control.setUser(who.ID, updated) s.auditControl(who, sn, "token.generate", nil, "ok", nil) // The plaintext token is returned exactly once — it is never stored, and the // status endpoint only ever exposes the last-4 hint afterwards. writeJSON(w, http.StatusOK, map[string]any{ "serial": sn, "token": token, "endpoint": s.ocppEndpoint(r, sn), }) } // handleAnkerControlRevoke deletes a charger's control token, so it can no longer // connect until a new token is generated. func (s *Server) handleAnkerControlRevoke(w http.ResponseWriter, r *http.Request) { who := caller(r) if who == nil { writeError(w, http.StatusUnauthorized, "not authenticated") return } if !s.pb.Configured() { writeError(w, http.StatusServiceUnavailable, "integration settings not configured on the server") return } sn := r.PathValue("sn") userRaw := s.userPluginSettings(r.Context(), who.ID) var updated map[string]ankerChargerBinding newDoc := mergeAnker(userRaw, func(as *ankerStored) { delete(as.ControlChargers, sn) updated = as.ControlChargers }) if err := s.pb.Update(r.Context(), s.usersCollection(), who.ID, map[string]any{"pluginSettings": newDoc}, nil); err != nil { writePBError(w, err) return } s.control.setUser(who.ID, updated) // Drop any live session for this charger so an already-connected charger using // the revoked token is disconnected too. if sess, ok := s.ocpp.SessionFor(sn); ok { sess.Close() } s.auditControl(who, sn, "token.revoke", nil, "ok", nil) writeJSON(w, http.StatusOK, map[string]any{"serial": sn, "revoked": true}) } // handleAnkerControlAction issues one OCPP command to a connected charger. func (s *Server) handleAnkerControlAction(w http.ResponseWriter, r *http.Request) { sess, ok := s.ankerControlSession(w, r) if !ok { return } who := caller(r) sn := r.PathValue("sn") action := r.PathValue("action") // Rate limit per user+charger so a valid session can't hammer the actuator. if !s.ctlRL.allow(who.ID + "|" + sn) { s.auditControl(who, sn, action, nil, "rate-limited", nil) writeError(w, http.StatusTooManyRequests, "too many control commands; please slow down") return } var body struct { IdTag string `json:"idTag"` ConnectorID int `json:"connectorId"` TransactionID int `json:"transactionId"` Amps float64 `json:"amps"` Operative *bool `json:"operative"` Hard bool `json:"hard"` RequestedMessage string `json:"requestedMessage"` Key string `json:"key"` Value string `json:"value"` Confirm bool `json:"confirm"` Password string `json:"password"` } if r.Body != nil { _ = json.NewDecoder(r.Body).Decode(&body) } // Destructive actions (reboot, cable unlock) demand an explicit confirmation // AND a fresh re-authentication (the caller re-enters their password), so they // can't fire by accident, from a replayed benign request, or from a session // left open on an unattended device. if isDestructiveAction(action) { if !body.Confirm { s.auditControl(who, sn, action, nil, "unconfirmed", nil) writeError(w, http.StatusBadRequest, "this action is destructive; resend with confirm:true") return } if !s.reauthenticate(r.Context(), who, body.Password) { s.auditControl(who, sn, action, nil, "reauth-failed", nil) writeError(w, http.StatusUnauthorized, "re-enter your password to confirm this action") return } } ctx := r.Context() var ( status string result any err error ) switch action { case "start": status, err = sess.RemoteStartTransaction(ctx, body.IdTag, body.ConnectorID) case "stop": status, err = sess.RemoteStopTransaction(ctx, body.TransactionID) case "limit": status, err = sess.SetCurrentLimit(ctx, body.ConnectorID, body.Amps) case "clear-limit": status, err = sess.ClearChargingProfile(ctx, body.ConnectorID) case "availability": operative := true if body.Operative != nil { operative = *body.Operative } status, err = sess.ChangeAvailability(ctx, body.ConnectorID, operative) case "reset": status, err = sess.Reset(ctx, body.Hard) case "unlock": status, err = sess.UnlockConnector(ctx, body.ConnectorID) case "trigger": status, err = sess.TriggerMessage(ctx, body.RequestedMessage, body.ConnectorID) case "config": if body.Key != "" { status, err = sess.ChangeConfiguration(ctx, body.Key, body.Value) } else { var conf ocpp.GetConfigurationResult conf, err = sess.GetConfiguration(ctx, nil) result = conf } default: writeError(w, http.StatusBadRequest, "unknown control action: "+action) return } // Audit the outcome (actor, charger, action, the meaningful params, result). outcome := status if err != nil { outcome = "error" } s.auditControl(who, sn, action, controlAuditParams(action, body.ConnectorID, body.Amps, body.Hard, body.Operative), outcome, err) if err != nil { writeJSON(w, http.StatusBadGateway, map[string]any{"error": err.Error()}) return } resp := map[string]any{"status": status} if result != nil { resp["result"] = result } writeJSON(w, http.StatusOK, resp) } // isDestructiveAction reports whether a control action reboots the charger or // releases the cable lock — actions that require an explicit confirm:true and a // password re-authentication. func isDestructiveAction(action string) bool { return action == "reset" || action == "unlock" } // reauthenticate verifies the caller's password against PocketBase (a sudo-style // step-up). It returns false on any failure — missing password, unknown email, // unconfigured service account, or wrong password. func (s *Server) reauthenticate(ctx context.Context, who *callerIdentity, password string) bool { if who == nil || who.Email == "" || password == "" || !s.pb.Configured() { return false } _, status, err := s.pb.LoginWithPassword(ctx, s.usersCollection(), who.Email, password) return err == nil && status == http.StatusOK } // controlAuditParams collects the parameters worth recording for an audited // control action (none are sensitive). func controlAuditParams(action string, connectorID int, amps float64, hard bool, operative *bool) map[string]any { p := map[string]any{} switch action { case "limit": p["amps"] = amps p["connectorId"] = connectorID case "reset": p["hard"] = hard case "availability": p["connectorId"] = connectorID if operative != nil { p["operative"] = *operative } case "start", "stop", "unlock", "clear-limit": if connectorID > 0 { p["connectorId"] = connectorID } } if len(p) == 0 { return nil } return p } // ankerControlSession applies the full gate (cascade + control mode + a live // session the caller actually owns) and returns the charger's session. func (s *Server) ankerControlSession(w http.ResponseWriter, r *http.Request) (*ocpp.Session, bool) { who := caller(r) if who == nil { writeError(w, http.StatusUnauthorized, "not authenticated") return nil, false } sn := r.PathValue("sn") userRaw := s.userPluginSettings(r.Context(), who.ID) res := s.resolveAnker(r.Context(), who, userRaw) switch { case !res.available: writeError(w, http.StatusForbidden, "the Anker Solix integration is disabled by the administrator") return nil, false case !res.orgEnabled: writeError(w, http.StatusForbidden, "the Anker Solix integration is disabled for your organization") return nil, false case !res.enabled: writeError(w, http.StatusForbidden, "enable the Anker Solix integration in Settings first") return nil, false } if res.eff.ControlMode == ankerControlOff { writeError(w, http.StatusConflict, "control mode is off; choose Own or Proxy CSMS to send commands") return nil, false } // The caller must own this charger (have a token bound to it), so a serial // alone can't be used to reach someone else's charger. if ankerBindingFor(userRaw, sn).TokenHash == "" { writeError(w, http.StatusNotFound, "no control token for this charger; generate one first") return nil, false } sess, ok := s.ocpp.SessionFor(sn) if !ok { writeError(w, http.StatusConflict, "charger is not connected to the control backend") return nil, false } return sess, true } // ---- helpers ----------------------------------------------------------------- // ocppEndpoint builds the ws(s):// URL the operator points the charger at. When // OCPP_PUBLIC_URL is configured it is used verbatim (scheme normalized to // ws/wss), which is safer than trusting request headers; otherwise it is derived // from the request. func (s *Server) ocppEndpoint(r *http.Request, sn string) string { s.mu.RLock() publicURL := s.cfg.OCPPPublicURL s.mu.RUnlock() if publicURL != "" { base := publicURL if rest, ok := strings.CutPrefix(base, "https://"); ok { base = "wss://" + rest } else if rest, ok := strings.CutPrefix(base, "http://"); ok { base = "ws://" + rest } return strings.TrimRight(base, "/") + "/ocpp/" + url.PathEscape(sn) } scheme := "ws" if requestIsSecure(r) { scheme = "wss" } return scheme + "://" + r.Host + "/ocpp/" + url.PathEscape(sn) } // requestIsSecure reports whether a request arrived over TLS, either directly or // via a TLS-terminating reverse proxy (X-Forwarded-Proto). The header is only // trustworthy behind such a proxy — the deployment model here (see the plan). func requestIsSecure(r *http.Request) bool { return r.TLS != nil || strings.EqualFold(r.Header.Get("X-Forwarded-Proto"), "https") } // ankerUpstreamAllowed reports whether a proxy-mode upstream URL points at an // Anker host, so a spoofed ocpp-info response can't steer the proxy elsewhere. func ankerUpstreamAllowed(raw string) bool { u, err := url.Parse(raw) if err != nil { return false } host := strings.ToLower(u.Hostname()) return host == "anker.com" || strings.HasSuffix(host, ".anker.com") } // ankerBindingFor reads one charger's stored control binding from a user's // pluginSettings blob. func ankerBindingFor(userRaw json.RawMessage, sn string) ankerChargerBinding { var doc ankerSettingsDoc if len(userRaw) > 0 { _ = json.Unmarshal(userRaw, &doc) } return doc.AnkerSolix.ControlChargers[sn] } func genControlToken() string { var b [24]byte _, _ = rand.Read(b[:]) return hex.EncodeToString(b[:]) } // hashToken returns the lowercase hex SHA-256 of a control token. Only this hash // is ever stored or indexed. func hashToken(token string) string { sum := sha256.Sum256([]byte(token)) return hex.EncodeToString(sum[:]) } // tokenHint returns the last 4 characters of a token, shown in the UI so an owner // can tell which token is installed without exposing it. func tokenHint(token string) string { if len(token) <= 4 { return token } return token[len(token)-4:] } // extractWSURL walks an arbitrary JSON value and returns the first ws:// or // wss:// string it finds. func extractWSURL(raw json.RawMessage) string { var v any if err := json.Unmarshal(raw, &v); err != nil { return "" } return walkForWS(v) } func walkForWS(v any) string { switch t := v.(type) { case string: s := strings.TrimSpace(t) if strings.HasPrefix(s, "ws://") || strings.HasPrefix(s, "wss://") { return s } case map[string]any: for _, val := range t { if u := walkForWS(val); u != "" { return u } } case []any: for _, val := range t { if u := walkForWS(val); u != "" { return u } } } return "" }