diff --git a/API Server/internal/api/integrations_ankersolix.go b/API Server/internal/api/integrations_ankersolix.go index 76a2993..3509c9f 100644 --- a/API Server/internal/api/integrations_ankersolix.go +++ b/API Server/internal/api/integrations_ankersolix.go @@ -27,13 +27,45 @@ import ( const ( ankerPlugin = "anker-solix" ankerSecretMask = "••••••••" + + // OCPP control modes for the Anker Solix charger (see internal/ocpp). + ankerControlOff = "off" // monitoring only (default) + ankerControlOwn = "own" // DriverVault is the charger's Central System + ankerControlProxy = "proxy" // DriverVault relays to Anker's cloud and injects ) +// normalizeControlMode maps a raw control-mode value to a recognized mode, or "" +// when unset/unknown so the cascade continues to the next layer. An explicit +// "off" is recognized (and thus wins its layer). +func normalizeControlMode(v string) string { + switch strings.ToLower(strings.TrimSpace(v)) { + case ankerControlOwn: + return ankerControlOwn + case ankerControlProxy: + return ankerControlProxy + case ankerControlOff: + return ankerControlOff + default: + return "" + } +} + // ankerConfig is one layer's Anker Solix settings. type ankerConfig struct { Email string `json:"email"` Password string `json:"password"` Country string `json:"country"` + // ControlMode is the OCPP control path: off | own | proxy (see internal/ocpp). + // It resolves independently of the credentials, like Country. + ControlMode string `json:"controlMode"` +} + +// ankerChargerBinding is the per-charger OCPP control token an operator installs +// into the charger (as the OCPP Basic-auth password) so it may connect to the +// DriverVault CSMS. It is user-layer only — not a cascade credential. +type ankerChargerBinding struct { + Token string `json:"token"` + AddedAt string `json:"addedAt,omitempty"` } // ankerStored is what we persist per user/org under pluginSettings.ankerSolix. @@ -44,6 +76,8 @@ type ankerStored struct { // Disabled is the organization layer's off switch, stored inverted so that // absent == enabled. Only meaningful on the org record; ignored on user records. Disabled bool `json:"disabled,omitempty"` + // ControlChargers maps a charger serial to its OCPP control token (user layer). + ControlChargers map[string]ankerChargerBinding `json:"controlChargers,omitempty"` } // ankerSettingsDoc is the pluginSettings JSON shape for the ankerSolix key. @@ -79,7 +113,7 @@ var ankerLayerRank = map[string]int{"global": 1, "org": 2, "user": 3} // pluginSettings blob (read from their user record). func (s *Server) resolveAnker(ctx context.Context, who *callerIdentity, userRaw json.RawMessage) ankerResolution { g, masterEnabled, _ := s.plugins.RawConfig(ankerPlugin) - gc := ankerConfig{Email: g["email"], Password: g["password"], Country: g["country"]} + gc := ankerConfig{Email: g["email"], Password: g["password"], Country: g["country"], ControlMode: g["controlMode"]} var oStored ankerStored if who.OrgID != "" { @@ -128,6 +162,18 @@ func (s *Server) resolveAnker(ctx context.Context, who *callerIdentity, userRaw } } + // Control mode resolves independently too, defaulting to off (monitoring + // only). The highest layer that sets a recognized value wins; an explicit + // "off" set above still wins (and locks lower layers to monitoring). + res.eff.ControlMode = ankerControlOff + res.source["controlMode"] = "unset" + for _, l := range layers { + if v := normalizeControlMode(l.c.ControlMode); v != "" { + res.eff.ControlMode, res.source["controlMode"] = v, l.name + break + } + } + // Credentials resolve as a pair from the highest layer with an email, so the // email and password never come from different layers. credSrc := "unset" @@ -234,9 +280,10 @@ func (s *Server) ankerScopeView(res ankerResolution, editable string) map[string return map[string]any{ "editableLayer": editable, "fields": map[string]ankerFieldView{ - "email": field("email", res.eff.Email, own.Email, false), - "password": field("password", res.eff.Password, own.Password, true), - "country": field("country", res.eff.Country, own.Country, false), + "email": field("email", res.eff.Email, own.Email, false), + "password": field("password", res.eff.Password, own.Password, true), + "country": field("country", res.eff.Country, own.Country, false), + "controlMode": field("controlMode", res.eff.ControlMode, own.ControlMode, false), }, } } @@ -247,6 +294,7 @@ func (s *Server) ankerView(who *callerIdentity, res ankerResolution) map[string] "available": res.available, "orgEnabled": res.orgEnabled, "enabled": res.enabled, + "controlMode": res.eff.ControlMode, // effective OCPP control mode (off|own|proxy) "role": who.Role, "orgId": who.OrgID, "canEditOrg": res.canOrg, @@ -340,6 +388,7 @@ func (s *Server) handlePutAnker(w http.ResponseWriter, r *http.Request) { applyField("email", func(c *ankerConfig, v string) { c.Email = v }) applyField("password", func(c *ankerConfig, v string) { c.Password = v }) applyField("country", func(c *ankerConfig, v string) { c.Country = strings.ToUpper(v) }) + applyField("controlMode", func(c *ankerConfig, v string) { c.ControlMode = normalizeControlMode(v) }) // Persist the organization layer (admins) via the service account. if editable == "org" { diff --git a/API Server/internal/api/integrations_ankersolix_control.go b/API Server/internal/api/integrations_ankersolix_control.go new file mode 100644 index 0000000..0f645fb --- /dev/null +++ b/API Server/internal/api/integrations_ankersolix_control.go @@ -0,0 +1,475 @@ +package api + +import ( + "context" + "crypto/rand" + "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 an OCPP control token to its owning user + charger. It is an +// in-memory cache, 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 + byToken map[string]ankerControlBinding + built bool +} + +func newControlIndex() *controlIndex { + return &controlIndex{byToken: map[string]ankerControlBinding{}} +} + +func (ci *controlIndex) lookup(token string) (ankerControlBinding, bool) { + ci.mu.RLock() + defer ci.mu.RUnlock() + b, ok := ci.byToken[token] + return b, ok +} + +// setUser replaces all of a user's token entries with the given charger set, so +// a regenerated token invalidates the previous one. +func (ci *controlIndex) setUser(userID string, chargers map[string]ankerChargerBinding) { + ci.mu.Lock() + defer ci.mu.Unlock() + for tok, b := range ci.byToken { + if b.UserID == userID { + delete(ci.byToken, tok) + } + } + for serial, cb := range chargers { + if cb.Token != "" { + ci.byToken[cb.Token] = 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() +} + +// ---- 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") + 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 + } + + 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() + } +} + +// 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") + } + 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.Token != "", + "token": binding.Token, // shown to the owner for charger provisioning + } + 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{Token: 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) + writeJSON(w, http.StatusOK, map[string]any{ + "serial": sn, + "token": token, + "endpoint": s.ocppEndpoint(r, sn), + }) +} + +// 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 + } + action := r.PathValue("action") + + 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"` + } + if r.Body != nil { + _ = json.NewDecoder(r.Body).Decode(&body) + } + + 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 + } + 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) +} + +// 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).Token == "" { + 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. +func (s *Server) ocppEndpoint(r *http.Request, sn string) string { + scheme := "ws" + if r.TLS != nil || strings.EqualFold(r.Header.Get("X-Forwarded-Proto"), "https") { + scheme = "wss" + } + return scheme + "://" + r.Host + "/ocpp/" + url.PathEscape(sn) +} + +// 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[:]) +} + +// 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 "" +} diff --git a/API Server/internal/api/integrations_ankersolix_test.go b/API Server/internal/api/integrations_ankersolix_test.go new file mode 100644 index 0000000..81dd668 --- /dev/null +++ b/API Server/internal/api/integrations_ankersolix_test.go @@ -0,0 +1,96 @@ +package api + +import ( + "encoding/json" + "net/http/httptest" + "testing" +) + +func TestNormalizeControlMode(t *testing.T) { + cases := map[string]string{ + "off": "off", + "own": "own", + "proxy": "proxy", + "OWN": "own", + " Proxy": "proxy", + "": "", // unset — cascade continues to the next layer + "bogus": "", // unknown — treated as unset + } + for in, want := range cases { + if got := normalizeControlMode(in); got != want { + t.Errorf("normalizeControlMode(%q) = %q, want %q", in, got, want) + } + } +} + +func TestControlIndexSetUserInvalidatesOldToken(t *testing.T) { + ci := newControlIndex() + ci.setUser("u1", map[string]ankerChargerBinding{"SN1": {Token: "tok-old"}}) + if _, ok := ci.lookup("tok-old"); !ok { + t.Fatal("tok-old should resolve after first set") + } + // Regenerate: same charger, new token. The old token must stop resolving. + ci.setUser("u1", map[string]ankerChargerBinding{"SN1": {Token: "tok-new"}}) + if _, ok := ci.lookup("tok-old"); ok { + t.Error("tok-old should be invalidated after regeneration") + } + b, ok := ci.lookup("tok-new") + if !ok || b.UserID != "u1" || b.Serial != "SN1" { + t.Errorf("tok-new resolves to %+v (ok=%v), want u1/SN1", b, ok) + } +} + +func TestControlIndexIsolatesUsers(t *testing.T) { + ci := newControlIndex() + ci.setUser("u1", map[string]ankerChargerBinding{"SN1": {Token: "a"}}) + ci.setUser("u2", map[string]ankerChargerBinding{"SN2": {Token: "b"}}) + // Re-setting u1 must not touch u2's token. + ci.setUser("u1", map[string]ankerChargerBinding{"SN1": {Token: "a2"}}) + if _, ok := ci.lookup("b"); !ok { + t.Error("u2's token should be untouched when u1 is updated") + } +} + +func TestAnkerBindingFor(t *testing.T) { + raw := json.RawMessage(`{"ankerSolix":{"controlChargers":{"SN1":{"token":"xyz"}}}}`) + if got := ankerBindingFor(raw, "SN1").Token; got != "xyz" { + t.Errorf("binding token = %q, want xyz", got) + } + if got := ankerBindingFor(raw, "other").Token; got != "" { + t.Errorf("unknown charger should have empty token, got %q", got) + } + if got := ankerBindingFor(nil, "SN1").Token; got != "" { + t.Errorf("nil settings should have empty token, got %q", got) + } +} + +func TestExtractWSURL(t *testing.T) { + cases := map[string]string{ + `{"data":{"ocppUrl":"wss://ocpp.anker.com/CP1"}}`: "wss://ocpp.anker.com/CP1", + `{"a":{"b":[{"endpoint":"ws://x/y"}]}}`: "ws://x/y", + `{"data":{"note":"https://not-a-ws-url"}}`: "", + `{"empty":true}`: "", + } + for in, want := range cases { + if got := extractWSURL(json.RawMessage(in)); got != want { + t.Errorf("extractWSURL(%s) = %q, want %q", in, got, want) + } + } +} + +func TestOCPPEndpoint(t *testing.T) { + s := &Server{} + // Plain HTTP request → ws:// + r := httptest.NewRequest("GET", "http://host.example/x", nil) + r.Host = "host.example" + if got := s.ocppEndpoint(r, "SN 1"); got != "ws://host.example/ocpp/SN%201" { + t.Errorf("endpoint = %q", got) + } + // Behind a TLS-terminating proxy → wss:// + r2 := httptest.NewRequest("GET", "http://host.example/x", nil) + r2.Host = "host.example" + r2.Header.Set("X-Forwarded-Proto", "https") + if got := s.ocppEndpoint(r2, "SN1"); got != "wss://host.example/ocpp/SN1" { + t.Errorf("tls endpoint = %q", got) + } +} diff --git a/API Server/internal/api/server.go b/API Server/internal/api/server.go index 6f32b25..77796cf 100644 --- a/API Server/internal/api/server.go +++ b/API Server/internal/api/server.go @@ -104,13 +104,16 @@ package api import ( + "bufio" "context" "log" + "net" "net/http" "sync" "time" "drivervault/apiserver/internal/config" + "drivervault/apiserver/internal/ocpp" "drivervault/apiserver/internal/pb" "drivervault/apiserver/internal/plugins" _ "drivervault/apiserver/internal/plugins/builtin" // register built-in plugins @@ -136,6 +139,13 @@ type Server struct { cfg config.Config pb *pb.Client plugins *plugins.Manager + + // ocpp is the OCPP 1.6J Central System that Anker Solix chargers connect to + // when their owner picks a control mode of own/proxy (see internal/ocpp and + // integrations_ankersolix_control.go). Nil-safe: control endpoints report a + // clear error when a charger is not connected. + ocpp *ocpp.CSMS + control *controlIndex // token -> owning user/charger for the /ocpp endpoint } // New constructs a Server around an already-built PocketBase client. @@ -144,14 +154,19 @@ func New(cfg config.Config, client *pb.Client) *Server { cfg: cfg, pb: client, plugins: plugins.NewManager(cfg.PluginsFile), + ocpp: ocpp.NewCSMS(func(f string, a ...any) { log.Printf("ocpp: "+f, a...) }), + control: newControlIndex(), } } // StartPlugins loads persisted plugin state and initialises enabled plugins. func (s *Server) StartPlugins() error { return s.plugins.Load() } -// Stop releases server-held resources (currently: plugin instances). -func (s *Server) Stop(ctx context.Context) { s.plugins.Shutdown(ctx) } +// Stop releases server-held resources (plugin instances and OCPP sessions). +func (s *Server) Stop(ctx context.Context) { + s.ocpp.Shutdown(ctx) + s.plugins.Shutdown(ctx) +} // usersCollection returns the PocketBase auth collection holding app users. func (s *Server) usersCollection() string { @@ -286,6 +301,18 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/integrations/anker-solix/health", s.handleAnkerHealth) mux.HandleFunc("GET /api/integrations/anker-solix/chargers", s.handleAnkerChargers) + // Anker Solix OCPP control (per-charger; gated by the same cascade plus a + // control mode of own/proxy and a live CSMS session). See + // integrations_ankersolix_control.go. + mux.HandleFunc("GET /api/integrations/anker-solix/chargers/{sn}/control", s.handleAnkerControlStatus) + mux.HandleFunc("POST /api/integrations/anker-solix/chargers/{sn}/control/token", s.handleAnkerControlToken) + mux.HandleFunc("POST /api/integrations/anker-solix/chargers/{sn}/{action}", s.handleAnkerControlAction) + + // OCPP WebSocket endpoint the charger dials out to (own/proxy modes). It sits + // outside /api/ so it bypasses bearer auth; it authenticates the charger with + // OCPP Basic auth (serial + per-charger control token) instead. + mux.HandleFunc("GET /ocpp/{serial}", s.handleOCPPConnect) + // Cars + sharing. mux.HandleFunc("GET /api/cars", s.listCars) mux.HandleFunc("POST /api/cars", s.createCar) @@ -454,3 +481,15 @@ func (w *statusWriter) Write(b []byte) (int, error) { w.wrote = true return w.ResponseWriter.Write(b) } + +// Hijack lets the wrapped ResponseWriter be taken over for a protocol switch +// (the OCPP WebSocket upgrade at /ocpp/{serial}). Without this pass-through the +// logging middleware would hide the underlying http.Hijacker. +func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { + hj, ok := w.ResponseWriter.(http.Hijacker) + if !ok { + return nil, nil, http.ErrNotSupported + } + w.wrote = true // a hijacked connection writes its own response + return hj.Hijack() +} diff --git a/API Server/internal/api/server_ocpp_test.go b/API Server/internal/api/server_ocpp_test.go new file mode 100644 index 0000000..cd65391 --- /dev/null +++ b/API Server/internal/api/server_ocpp_test.go @@ -0,0 +1,63 @@ +package api + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "drivervault/apiserver/internal/config" + "drivervault/apiserver/internal/ocpp" + "drivervault/apiserver/internal/pb" +) + +// TestOCPPRouteAndControlAuth exercises the real HTTP stack (all middleware +// included) to confirm the OCPP endpoint and the control REST routes are wired, +// and that the charger-auth gate rejects an unknown token. PocketBase is left +// unconfigured, so the successful (charger-connected) path is out of scope here — +// that is covered by the internal/ocpp own/proxy tests. +func TestOCPPRouteAndControlAuth(t *testing.T) { + s := New(config.Config{UsersCollection: "users"}, pb.New("", "", "")) + srv := httptest.NewServer(s.Handler()) + defer srv.Close() + + wsBase := "ws" + strings.TrimPrefix(srv.URL, "http") + + // A charger connecting with no OCPP Basic auth is rejected (401). + if _, err := ocpp.Dial(context.Background(), wsBase+"/ocpp/CP1", []string{"ocpp1.6"}, nil); err == nil || + !strings.Contains(err.Error(), "401") { + t.Fatalf("unauthenticated charger connect: want HTTP 401, got %v", err) + } + + // A bogus token is unknown to the index (PB unconfigured, so the rebuild is a + // no-op) and is likewise rejected. + h := http.Header{} + h.Set("Authorization", ocpp.BasicAuthHeader("CP1", "bogus-token")) + if _, err := ocpp.Dial(context.Background(), wsBase+"/ocpp/CP1", []string{"ocpp1.6"}, h); err == nil || + !strings.Contains(err.Error(), "401") { + t.Fatalf("bogus-token charger connect: want HTTP 401, got %v", err) + } + + // The control REST routes are registered (401 for a missing bearer token, not + // 404 for an unknown route) — this also proves the {sn}/control, + // {sn}/control/token and {sn}/{action} patterns don't shadow each other. + routes := []struct { + method, path string + }{ + {http.MethodGet, "/api/integrations/anker-solix/chargers/CP1/control"}, + {http.MethodPost, "/api/integrations/anker-solix/chargers/CP1/control/token"}, + {http.MethodPost, "/api/integrations/anker-solix/chargers/CP1/start"}, + } + for _, rt := range routes { + req, _ := http.NewRequest(rt.method, srv.URL+rt.path, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s %s: %v", rt.method, rt.path, err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Errorf("%s %s: status %d, want 401 (route registered, bearer required)", rt.method, rt.path, resp.StatusCode) + } + } +} diff --git a/API Server/internal/ocpp/commands.go b/API Server/internal/ocpp/commands.go new file mode 100644 index 0000000..409eda3 --- /dev/null +++ b/API Server/internal/ocpp/commands.go @@ -0,0 +1,164 @@ +package ocpp + +import ( + "context" + "encoding/json" + "errors" +) + +// This file is the outbound control surface: typed helpers that build an OCPP +// CALL, send it to the charger via Session.Call, and parse the reply. They work +// identically in own and proxy modes (both inject over the same socket). + +// StatusResponse is the common {status: "..."} reply most control calls return +// (e.g. "Accepted", "Rejected", "Scheduled", "NotSupported"). +type StatusResponse struct { + Status string `json:"status"` +} + +// statusCall issues a control CALL that replies with a {status} object. +func (s *Session) statusCall(ctx context.Context, action string, payload any) (string, error) { + raw, err := s.Call(ctx, action, payload) + if err != nil { + return "", err + } + var r StatusResponse + if err := json.Unmarshal(raw, &r); err != nil { + return "", err + } + return r.Status, nil +} + +// RemoteStartTransaction asks the charger to begin a charging session. A blank +// idTag is defaulted; connectorID <= 0 lets the charger choose. +func (s *Session) RemoteStartTransaction(ctx context.Context, idTag string, connectorID int) (string, error) { + if idTag == "" { + idTag = "DriverVault" + } + payload := map[string]any{"idTag": idTag} + if connectorID > 0 { + payload["connectorId"] = connectorID + } + return s.statusCall(ctx, "RemoteStartTransaction", payload) +} + +// RemoteStopTransaction stops a session. When transactionID <= 0 the id from the +// live status snapshot (own mode) is used. +func (s *Session) RemoteStopTransaction(ctx context.Context, transactionID int) (string, error) { + if transactionID <= 0 { + transactionID = s.Snapshot().TransactionID + } + if transactionID <= 0 { + return "", errors.New("ocpp: no active transaction to stop") + } + return s.statusCall(ctx, "RemoteStopTransaction", map[string]any{"transactionId": transactionID}) +} + +// SetCurrentLimit caps the charge current (Amperes) via a TxDefaultProfile. A +// connectorID <= 0 defaults to connector 1. +func (s *Session) SetCurrentLimit(ctx context.Context, connectorID int, amps float64) (string, error) { + if connectorID <= 0 { + connectorID = 1 + } + profile := map[string]any{ + "chargingProfileId": 1, + "stackLevel": 0, + "chargingProfilePurpose": "TxDefaultProfile", + "chargingProfileKind": "Relative", + "chargingSchedule": map[string]any{ + "chargingRateUnit": "A", + "chargingSchedulePeriod": []any{ + map[string]any{"startPeriod": 0, "limit": amps}, + }, + }, + } + return s.statusCall(ctx, "SetChargingProfile", map[string]any{ + "connectorId": connectorID, + "csChargingProfiles": profile, + }) +} + +// ClearChargingProfile removes charging profiles (lifting a current limit). A +// connectorID <= 0 clears all connectors. +func (s *Session) ClearChargingProfile(ctx context.Context, connectorID int) (string, error) { + payload := map[string]any{} + if connectorID > 0 { + payload["connectorId"] = connectorID + } + return s.statusCall(ctx, "ClearChargingProfile", payload) +} + +// ChangeAvailability sets a connector Operative/Inoperative. connectorID 0 +// targets the whole charge point. +func (s *Session) ChangeAvailability(ctx context.Context, connectorID int, operative bool) (string, error) { + typ := "Inoperative" + if operative { + typ = "Operative" + } + return s.statusCall(ctx, "ChangeAvailability", map[string]any{ + "connectorId": connectorID, + "type": typ, + }) +} + +// Reset reboots the charger (Soft or Hard). +func (s *Session) Reset(ctx context.Context, hard bool) (string, error) { + typ := "Soft" + if hard { + typ = "Hard" + } + return s.statusCall(ctx, "Reset", map[string]any{"type": typ}) +} + +// UnlockConnector releases the cable lock on a connector. +func (s *Session) UnlockConnector(ctx context.Context, connectorID int) (string, error) { + if connectorID <= 0 { + connectorID = 1 + } + return s.statusCall(ctx, "UnlockConnector", map[string]any{"connectorId": connectorID}) +} + +// TriggerMessage asks the charger to proactively send a message (e.g. +// "StatusNotification", "MeterValues", "BootNotification", "Heartbeat"). +func (s *Session) TriggerMessage(ctx context.Context, requestedMessage string, connectorID int) (string, error) { + payload := map[string]any{"requestedMessage": requestedMessage} + if connectorID > 0 { + payload["connectorId"] = connectorID + } + return s.statusCall(ctx, "TriggerMessage", payload) +} + +// ChangeConfiguration sets a charger configuration key. +func (s *Session) ChangeConfiguration(ctx context.Context, key, value string) (string, error) { + return s.statusCall(ctx, "ChangeConfiguration", map[string]any{"key": key, "value": value}) +} + +// ConfigKey is one entry in a GetConfiguration reply. +type ConfigKey struct { + Key string `json:"key"` + Readonly bool `json:"readonly"` + Value string `json:"value,omitempty"` +} + +// GetConfigurationResult is the GetConfiguration reply. +type GetConfigurationResult struct { + ConfigurationKey []ConfigKey `json:"configurationKey"` + UnknownKey []string `json:"unknownKey,omitempty"` +} + +// GetConfiguration reads charger configuration keys; nil/empty keys returns all. +func (s *Session) GetConfiguration(ctx context.Context, keys []string) (GetConfigurationResult, error) { + payload := map[string]any{} + if len(keys) > 0 { + payload["key"] = keys + } + raw, err := s.Call(ctx, "GetConfiguration", payload) + if err != nil { + return GetConfigurationResult{}, err + } + var r GetConfigurationResult + if err := json.Unmarshal(raw, &r); err != nil { + return GetConfigurationResult{}, err + } + return r, nil +} diff --git a/API Server/internal/ocpp/csms.go b/API Server/internal/ocpp/csms.go new file mode 100644 index 0000000..ccb94d4 --- /dev/null +++ b/API Server/internal/ocpp/csms.go @@ -0,0 +1,146 @@ +package ocpp + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" +) + +// CSMS is the Central System: it holds the live charge-point sessions keyed by +// serial and answers the inbound OCPP calls a charger makes in own mode. +type CSMS struct { + mu sync.RWMutex + sessions map[string]*Session + logf func(string, ...any) +} + +// NewCSMS builds an empty Central System. logf may be nil. +func NewCSMS(logf func(string, ...any)) *CSMS { + if logf == nil { + logf = func(string, ...any) {} + } + return &CSMS{sessions: map[string]*Session{}, logf: logf} +} + +// Accept registers a newly-connected charger. In own mode cp is the charger +// connection and up must be nil; in proxy mode up is the (already-dialed) +// upstream CSMS connection. Any prior session for the same serial is closed. The +// returned Session is where control commands are issued. +func (c *CSMS) Accept(serial, mode string, cp, up *Conn) (*Session, error) { + if serial == "" { + return nil, errors.New("ocpp: empty serial") + } + switch mode { + case ModeOwn: + if up != nil { + _ = up.Close() + up = nil + } + case ModeProxy: + if up == nil { + return nil, errors.New("ocpp: proxy mode requires an upstream connection") + } + default: + return nil, fmt.Errorf("ocpp: cannot accept in mode %q", mode) + } + + // Evict any existing session for this serial (a reconnect) before inserting. + c.mu.Lock() + old := c.sessions[serial] + delete(c.sessions, serial) + c.mu.Unlock() + if old != nil { + old.close(nil) + } + + sess := newSession(serial, mode, cp, up, c.handleCall, c.deregister, c.logf) + c.mu.Lock() + c.sessions[serial] = sess + c.mu.Unlock() + sess.start() + return sess, nil +} + +// SessionFor returns the live session for a serial, if the charger is connected. +func (c *CSMS) SessionFor(serial string) (*Session, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + s, ok := c.sessions[serial] + return s, ok +} + +// Statuses returns a snapshot of every connected charger. +func (c *CSMS) Statuses() []Status { + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]Status, 0, len(c.sessions)) + for _, s := range c.sessions { + out = append(out, s.Snapshot()) + } + return out +} + +// Shutdown closes every session. +func (c *CSMS) Shutdown(context.Context) { + c.mu.Lock() + sessions := make([]*Session, 0, len(c.sessions)) + for _, s := range c.sessions { + sessions = append(sessions, s) + } + c.sessions = map[string]*Session{} + c.mu.Unlock() + for _, s := range sessions { + s.close(nil) + } +} + +func (c *CSMS) deregister(s *Session) { + c.mu.Lock() + if c.sessions[s.serial] == s { + delete(c.sessions, s.serial) + } + c.mu.Unlock() +} + +// handleCall answers an inbound CALL from a charger in own mode. It implements +// the CSMS side of the OCPP 1.6 core profile: enough for a charger to boot, +// heartbeat, report status/meter values and open/close transactions against us. +func (c *CSMS) handleCall(s *Session, action string, payload json.RawMessage) (any, string, string) { + now := time.Now().UTC().Format(time.RFC3339) + switch action { + case "BootNotification": + return map[string]any{"status": "Accepted", "currentTime": now, "interval": 300}, "", "" + case "Heartbeat": + return map[string]any{"currentTime": now}, "", "" + case "StatusNotification", "MeterValues", + "FirmwareStatusNotification", "DiagnosticsStatusNotification": + return map[string]any{}, "", "" + case "Authorize": + return map[string]any{"idTagInfo": map[string]any{"status": "Accepted"}}, "", "" + case "StartTransaction": + return map[string]any{ + "transactionId": s.assignTxn(), + "idTagInfo": map[string]any{"status": "Accepted"}, + }, "", "" + case "StopTransaction": + return map[string]any{"idTagInfo": map[string]any{"status": "Accepted"}}, "", "" + case "DataTransfer": + return map[string]any{"status": "Accepted"}, "", "" + default: + return nil, ErrNotImplemented, "action not supported by DriverVault CSMS" + } +} + +// assignTxn allocates a transaction id in own mode and records it in the status. +func (s *Session) assignTxn() int { + s.mu.Lock() + s.nextTxn++ + txn := s.nextTxn + s.status.TransactionID = txn + s.status.LastUpdated = time.Now() + s.mu.Unlock() + return txn +} diff --git a/API Server/internal/ocpp/csms_test.go b/API Server/internal/ocpp/csms_test.go new file mode 100644 index 0000000..2870b71 --- /dev/null +++ b/API Server/internal/ocpp/csms_test.go @@ -0,0 +1,147 @@ +package ocpp + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path" + "testing" + "time" +) + +// ownModeServer stands up a CSMS behind an httptest server that upgrades every +// /ocpp/{serial} connection and registers it in own mode. +func ownModeServer(t *testing.T) (*CSMS, *httptest.Server, <-chan *Session) { + t.Helper() + csms := NewCSMS(nil) + sessCh := make(chan *Session, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := Upgrade(w, r) + if err != nil { + t.Errorf("upgrade: %v", err) + return + } + sess, err := csms.Accept(path.Base(r.URL.Path), ModeOwn, conn, nil) + if err != nil { + t.Errorf("accept: %v", err) + _ = conn.Close() + return + } + sessCh <- sess + })) + return csms, srv, sessCh +} + +func TestOwnModeEndToEnd(t *testing.T) { + csms, srv, sessCh := ownModeServer(t) + defer srv.Close() + + cp := dialSimCP(t, wsURL(srv.URL, "/ocpp/CP-A5191"), nil) + defer cp.close() + ctx := context.Background() + + // Boot: the CSMS accepts and the charger appears connected. + boot := cp.call(t, "BootNotification", map[string]any{ + "chargePointVendor": "Anker", "chargePointModel": "A5191", "firmwareVersion": "1.2.3", + }) + assertStatus(t, boot, "Accepted") + + var sess *Session + select { + case sess = <-sessCh: + case <-time.After(2 * time.Second): + t.Fatal("CSMS never registered the session") + } + if _, ok := csms.SessionFor("CP-A5191"); !ok { + t.Fatal("SessionFor should find the connected charger") + } + + // StatusNotification + MeterValues update the tapped snapshot. + cp.call(t, "StatusNotification", map[string]any{"connectorId": 1, "status": "Charging", "errorCode": "NoError"}) + cp.call(t, "MeterValues", map[string]any{ + "connectorId": 1, + "meterValue": []any{map[string]any{ + "timestamp": time.Now().UTC().Format(time.RFC3339), + "sampledValue": []any{map[string]any{"value": "1500", "measurand": "Energy.Active.Import.Register", "unit": "Wh"}}, + }}, + }) + snap := sess.Snapshot() + if snap.Vendor != "Anker" || snap.Firmware != "1.2.3" { + t.Errorf("boot fields not tapped: %+v", snap) + } + if snap.ConnectorStatus != "Charging" { + t.Errorf("connector status = %q, want Charging", snap.ConnectorStatus) + } + if snap.MeterWh != 1500 { + t.Errorf("meter = %d Wh, want 1500", snap.MeterWh) + } + + // StartTransaction: the CSMS assigns a transaction id. + st := cp.call(t, "StartTransaction", map[string]any{"connectorId": 1, "idTag": "TAG", "meterStart": 0, "timestamp": time.Now().UTC().Format(time.RFC3339)}) + var startResp struct { + TransactionID int `json:"transactionId"` + } + _ = json.Unmarshal(st.Payload, &startResp) + if startResp.TransactionID == 0 { + t.Fatal("StartTransaction should return a non-zero transaction id") + } + + // Now drive every control command and confirm the charger sees it + Accepts. + cases := []struct { + name string + run func() (string, error) + action string + }{ + {"RemoteStart", func() (string, error) { return sess.RemoteStartTransaction(ctx, "TAG", 1) }, "RemoteStartTransaction"}, + {"SetCurrentLimit", func() (string, error) { return sess.SetCurrentLimit(ctx, 1, 16) }, "SetChargingProfile"}, + {"ClearChargingProfile", func() (string, error) { return sess.ClearChargingProfile(ctx, 1) }, "ClearChargingProfile"}, + {"ChangeAvailability", func() (string, error) { return sess.ChangeAvailability(ctx, 0, false) }, "ChangeAvailability"}, + {"UnlockConnector", func() (string, error) { return sess.UnlockConnector(ctx, 1) }, "UnlockConnector"}, + {"TriggerMessage", func() (string, error) { return sess.TriggerMessage(ctx, "StatusNotification", 1) }, "TriggerMessage"}, + {"ChangeConfiguration", func() (string, error) { return sess.ChangeConfiguration(ctx, "HeartbeatInterval", "60") }, "ChangeConfiguration"}, + {"Reset", func() (string, error) { return sess.Reset(ctx, false) }, "Reset"}, + {"RemoteStop", func() (string, error) { return sess.RemoteStopTransaction(ctx, startResp.TransactionID) }, "RemoteStopTransaction"}, + } + for _, c := range cases { + status, err := c.run() + if err != nil { + t.Fatalf("%s: %v", c.name, err) + } + if status != "Accepted" { + t.Errorf("%s: status = %q, want Accepted", c.name, status) + } + got := cp.waitRecv(t) + if got.Action != c.action { + t.Errorf("%s: charger received %q, want %q", c.name, got.Action, c.action) + } + } + + // GetConfiguration returns the charger's keys. + conf, err := sess.GetConfiguration(ctx, nil) + if err != nil { + t.Fatalf("GetConfiguration: %v", err) + } + cp.waitRecv(t) + if len(conf.ConfigurationKey) == 0 || conf.ConfigurationKey[0].Key != "HeartbeatInterval" { + t.Errorf("GetConfiguration = %+v", conf) + } +} + +func TestSessionForUnknownCharger(t *testing.T) { + csms := NewCSMS(nil) + if _, ok := csms.SessionFor("nope"); ok { + t.Fatal("SessionFor should be false for an unconnected charger") + } +} + +func assertStatus(t *testing.T, m Message, want string) { + t.Helper() + var r StatusResponse + if err := json.Unmarshal(m.Payload, &r); err != nil { + t.Fatalf("decode status: %v", err) + } + if r.Status != want { + t.Fatalf("status = %q, want %q", r.Status, want) + } +} diff --git a/API Server/internal/ocpp/helpers_test.go b/API Server/internal/ocpp/helpers_test.go new file mode 100644 index 0000000..aed0dae --- /dev/null +++ b/API Server/internal/ocpp/helpers_test.go @@ -0,0 +1,125 @@ +package ocpp + +import ( + "context" + "net/http" + "strings" + "sync" + "testing" + "time" +) + +// wsURL turns an httptest http:// base URL into a ws:// URL with a path. +func wsURL(base, path string) string { + return "ws" + strings.TrimPrefix(base, "http") + path +} + +// simCP is a simulated OCPP 1.6J charge point used by the CSMS and proxy tests. +// It auto-answers inbound control CALLs with "Accepted" (recording them for +// assertions) and can itself send CALLs (BootNotification, StatusNotification, …) +// and await their result. +type simCP struct { + conn *Conn + + mu sync.Mutex + pending map[string]chan Message + + recv chan Message // inbound CALLs (CSMS → CP), for assertions +} + +func dialSimCP(t *testing.T, url string, header http.Header) *simCP { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + conn, err := Dial(ctx, url, []string{"ocpp1.6"}, header) + if err != nil { + t.Fatalf("dial sim charge point: %v", err) + } + cp := &simCP{conn: conn, pending: map[string]chan Message{}, recv: make(chan Message, 64)} + go cp.loop() + return cp +} + +func (cp *simCP) loop() { + for { + data, err := cp.conn.ReadMessage() + if err != nil { + return + } + msg, err := DecodeMessage(data) + if err != nil { + continue + } + switch msg.Type { + case MessageTypeCall: + select { + case cp.recv <- msg: + default: + } + out, _ := EncodeCallResult(msg.ID, cp.responseFor(msg.Action)) + _ = cp.conn.WriteMessage(out) + case MessageTypeCallResult, MessageTypeCallError: + cp.mu.Lock() + ch := cp.pending[msg.ID] + delete(cp.pending, msg.ID) + cp.mu.Unlock() + if ch != nil { + ch <- msg + } + } + } +} + +// responseFor returns the payload the sim answers an inbound control CALL with. +func (cp *simCP) responseFor(action string) any { + switch action { + case "GetConfiguration": + return map[string]any{ + "configurationKey": []any{ + map[string]any{"key": "HeartbeatInterval", "readonly": false, "value": "300"}, + }, + } + default: + // RemoteStart/Stop, Reset, ChangeAvailability, UnlockConnector, + // TriggerMessage, SetChargingProfile, ClearChargingProfile, ChangeConfiguration. + return map[string]any{"status": "Accepted"} + } +} + +// call sends a CALL from the charge point and waits for the reply. +func (cp *simCP) call(t *testing.T, action string, payload any) Message { + t.Helper() + id := newMessageID() + frame, err := EncodeCall(id, action, payload) + if err != nil { + t.Fatalf("encode %s: %v", action, err) + } + ch := make(chan Message, 1) + cp.mu.Lock() + cp.pending[id] = ch + cp.mu.Unlock() + if err := cp.conn.WriteMessage(frame); err != nil { + t.Fatalf("write %s: %v", action, err) + } + select { + case m := <-ch: + return m + case <-time.After(5 * time.Second): + t.Fatalf("timeout awaiting %s reply", action) + return Message{} + } +} + +// waitRecv returns the next inbound control CALL the sim received, or fails. +func (cp *simCP) waitRecv(t *testing.T) Message { + t.Helper() + select { + case m := <-cp.recv: + return m + case <-time.After(5 * time.Second): + t.Fatal("timeout awaiting inbound control call") + return Message{} + } +} + +func (cp *simCP) close() { _ = cp.conn.Close() } diff --git a/API Server/internal/ocpp/ocpp16.go b/API Server/internal/ocpp/ocpp16.go new file mode 100644 index 0000000..d514d2a --- /dev/null +++ b/API Server/internal/ocpp/ocpp16.go @@ -0,0 +1,142 @@ +package ocpp + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" +) + +// OCPP-J message types (OCPP 1.6 §4.2). Every frame is a JSON array whose first +// element is one of these. +const ( + MessageTypeCall = 2 // [2, id, action, payload] + MessageTypeCallResult = 3 // [3, id, payload] + MessageTypeCallError = 4 // [4, id, errorCode, errorDescription, errorDetails] +) + +// Standard OCPP-J error codes used when rejecting a CALL. +const ( + ErrNotImplemented = "NotImplemented" + ErrNotSupported = "NotSupported" + ErrInternalError = "InternalError" + ErrProtocolError = "ProtocolError" + ErrSecurityError = "SecurityError" + ErrFormationViolation = "FormationViolation" + ErrPropertyConstraintViolation = "PropertyConstraintViolation" + ErrGenericError = "GenericError" +) + +// Message is a decoded OCPP-J frame; only the fields relevant to Type are set. +type Message struct { + Type int + ID string + Action string // CALL only + Payload json.RawMessage // CALL and CALLRESULT + + ErrorCode string // CALLERROR only + ErrorDescription string // CALLERROR only + ErrorDetails json.RawMessage // CALLERROR only +} + +// DecodeMessage parses one OCPP-J frame. +func DecodeMessage(b []byte) (Message, error) { + var arr []json.RawMessage + if err := json.Unmarshal(b, &arr); err != nil { + return Message{}, fmt.Errorf("ocpp: not a JSON array: %w", err) + } + if len(arr) < 3 { + return Message{}, errors.New("ocpp: message array too short") + } + var typ int + if err := json.Unmarshal(arr[0], &typ); err != nil { + return Message{}, fmt.Errorf("ocpp: bad message type: %w", err) + } + var id string + if err := json.Unmarshal(arr[1], &id); err != nil { + return Message{}, fmt.Errorf("ocpp: bad message id: %w", err) + } + m := Message{Type: typ, ID: id} + + switch typ { + case MessageTypeCall: + if len(arr) != 4 { + return Message{}, errors.New("ocpp: CALL must have 4 elements") + } + if err := json.Unmarshal(arr[2], &m.Action); err != nil { + return Message{}, fmt.Errorf("ocpp: bad action: %w", err) + } + m.Payload = arr[3] + case MessageTypeCallResult: + m.Payload = arr[2] + case MessageTypeCallError: + if len(arr) != 5 { + return Message{}, errors.New("ocpp: CALLERROR must have 5 elements") + } + _ = json.Unmarshal(arr[2], &m.ErrorCode) + _ = json.Unmarshal(arr[3], &m.ErrorDescription) + m.ErrorDetails = arr[4] + default: + return Message{}, fmt.Errorf("ocpp: unknown message type %d", typ) + } + return m, nil +} + +// EncodeCall builds a CALL frame. A nil/empty payload is encoded as {} because +// OCPP requires the payload to be a JSON object. +func EncodeCall(id, action string, payload any) ([]byte, error) { + p, err := payloadObject(payload) + if err != nil { + return nil, err + } + return json.Marshal([]any{MessageTypeCall, id, action, p}) +} + +// EncodeCallResult builds a CALLRESULT frame answering the CALL with id. +func EncodeCallResult(id string, payload any) ([]byte, error) { + p, err := payloadObject(payload) + if err != nil { + return nil, err + } + return json.Marshal([]any{MessageTypeCallResult, id, p}) +} + +// EncodeCallError builds a CALLERROR frame rejecting the CALL with id. +func EncodeCallError(id, code, description string, details any) ([]byte, error) { + d, err := payloadObject(details) + if err != nil { + return nil, err + } + return json.Marshal([]any{MessageTypeCallError, id, code, description, d}) +} + +// payloadObject normalizes any payload to a JSON object RawMessage, mapping +// nil/null to the empty object {}. +func payloadObject(payload any) (json.RawMessage, error) { + switch v := payload.(type) { + case nil: + return json.RawMessage("{}"), nil + case json.RawMessage: + if len(v) == 0 || string(v) == "null" { + return json.RawMessage("{}"), nil + } + return v, nil + default: + b, err := json.Marshal(payload) + if err != nil { + return nil, err + } + if len(b) == 0 || string(b) == "null" { + return json.RawMessage("{}"), nil + } + return b, nil + } +} + +// newMessageID returns a fresh unique id for an outbound CALL. +func newMessageID() string { + var b [8]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} diff --git a/API Server/internal/ocpp/ocpp16_test.go b/API Server/internal/ocpp/ocpp16_test.go new file mode 100644 index 0000000..a300ab2 --- /dev/null +++ b/API Server/internal/ocpp/ocpp16_test.go @@ -0,0 +1,64 @@ +package ocpp + +import ( + "encoding/json" + "testing" +) + +func TestDecodeCall(t *testing.T) { + m, err := DecodeMessage([]byte(`[2,"abc","BootNotification",{"chargePointModel":"A5191"}]`)) + if err != nil { + t.Fatal(err) + } + if m.Type != MessageTypeCall || m.ID != "abc" || m.Action != "BootNotification" { + t.Fatalf("bad decode: %+v", m) + } + var p struct { + Model string `json:"chargePointModel"` + } + if err := json.Unmarshal(m.Payload, &p); err != nil || p.Model != "A5191" { + t.Fatalf("payload = %s (%v)", m.Payload, err) + } +} + +func TestDecodeCallResultAndError(t *testing.T) { + r, err := DecodeMessage([]byte(`[3,"abc",{"status":"Accepted"}]`)) + if err != nil || r.Type != MessageTypeCallResult { + t.Fatalf("callresult decode: %+v %v", r, err) + } + e, err := DecodeMessage([]byte(`[4,"abc","NotImplemented","nope",{}]`)) + if err != nil || e.Type != MessageTypeCallError || e.ErrorCode != "NotImplemented" || e.ErrorDescription != "nope" { + t.Fatalf("callerror decode: %+v %v", e, err) + } +} + +func TestEncodeRoundTrip(t *testing.T) { + b, err := EncodeCall("id1", "Reset", map[string]any{"type": "Soft"}) + if err != nil { + t.Fatal(err) + } + m, err := DecodeMessage(b) + if err != nil || m.Action != "Reset" { + t.Fatalf("round trip: %+v %v", m, err) + } +} + +func TestEncodeEmptyPayloadIsObject(t *testing.T) { + // OCPP requires the payload be a JSON object, never null. + b, err := EncodeCallResult("id1", nil) + if err != nil { + t.Fatal(err) + } + if string(b) != `[3,"id1",{}]` { + t.Fatalf("empty payload encoded as %s, want [3,\"id1\",{}]", b) + } +} + +func TestDecodeRejectsGarbage(t *testing.T) { + if _, err := DecodeMessage([]byte(`{"not":"an array"}`)); err == nil { + t.Fatal("expected error decoding non-array") + } + if _, err := DecodeMessage([]byte(`[2,"id"]`)); err == nil { + t.Fatal("expected error decoding short CALL") + } +} diff --git a/API Server/internal/ocpp/proxy.go b/API Server/internal/ocpp/proxy.go new file mode 100644 index 0000000..874e6fd --- /dev/null +++ b/API Server/internal/ocpp/proxy.go @@ -0,0 +1,28 @@ +package ocpp + +import ( + "context" + "encoding/base64" + "net/http" +) + +// Proxy mode reuses the Session read loops (session.go) to pump frames between +// the charger and Anker's cloud while tapping status and injecting our own +// control calls. This file only provides the upstream dialer and auth helpers; +// the CSMS is handed the already-dialed upstream connection via Accept. + +// DialUpstream opens a client OCPP 1.6J connection to an upstream CSMS (Anker's +// cloud) for proxy mode. authHeader, when non-empty, is sent as the Authorization +// request header (OCPP security profile 1 Basic auth). +func DialUpstream(ctx context.Context, url, authHeader string) (*Conn, error) { + h := http.Header{} + if authHeader != "" { + h.Set("Authorization", authHeader) + } + return Dial(ctx, url, []string{"ocpp1.6"}, h) +} + +// BasicAuthHeader builds an HTTP Basic Authorization header value. +func BasicAuthHeader(user, pass string) string { + return "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+pass)) +} diff --git a/API Server/internal/ocpp/proxy_test.go b/API Server/internal/ocpp/proxy_test.go new file mode 100644 index 0000000..33ad561 --- /dev/null +++ b/API Server/internal/ocpp/proxy_test.go @@ -0,0 +1,129 @@ +package ocpp + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// fakeUpstream is a stand-in for Anker's cloud CSMS: it upgrades, records every +// CALL it receives, and answers each with {status:"Accepted"} (plus boot fields). +func fakeUpstream(t *testing.T) (*httptest.Server, <-chan Message) { + t.Helper() + recv := make(chan Message, 64) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := Upgrade(w, r) + if err != nil { + t.Errorf("upstream upgrade: %v", err) + return + } + for { + data, err := conn.ReadMessage() + if err != nil { + return + } + msg, err := DecodeMessage(data) + if err != nil { + continue + } + if msg.Type != MessageTypeCall { + continue + } + select { + case recv <- msg: + default: + } + out, _ := EncodeCallResult(msg.ID, map[string]any{ + "status": "Accepted", "currentTime": time.Now().UTC().Format(time.RFC3339), "interval": 300, + }) + _ = conn.WriteMessage(out) + } + })) + return srv, recv +} + +func TestProxyModeForwardsAndInjects(t *testing.T) { + upSrv, upRecv := fakeUpstream(t) + defer upSrv.Close() + upURL := wsURL(upSrv.URL, "/ocpp/CP-A5191") + + csms := NewCSMS(nil) + sessCh := make(chan *Session, 1) + dvSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cpConn, err := Upgrade(w, r) + if err != nil { + t.Errorf("dv upgrade: %v", err) + return + } + upConn, err := DialUpstream(context.Background(), upURL, BasicAuthHeader("CP-A5191", "secret")) + if err != nil { + t.Errorf("dial upstream: %v", err) + _ = cpConn.Close() + return + } + sess, err := csms.Accept("CP-A5191", ModeProxy, cpConn, upConn) + if err != nil { + t.Errorf("accept proxy: %v", err) + return + } + sessCh <- sess + })) + defer dvSrv.Close() + + cp := dialSimCP(t, wsURL(dvSrv.URL, "/ocpp/CP-A5191"), nil) + defer cp.close() + + // Charger→upstream: BootNotification is forwarded and the upstream's answer + // comes back to the charger. + boot := cp.call(t, "BootNotification", map[string]any{"chargePointModel": "A5191"}) + assertStatus(t, boot, "Accepted") + if got := waitMsg(t, upRecv); got.Action != "BootNotification" { + t.Fatalf("upstream received %q, want BootNotification", got.Action) + } + + var sess *Session + select { + case sess = <-sessCh: + case <-time.After(2 * time.Second): + t.Fatal("proxy session never registered") + } + + // A forwarded StatusNotification is tapped for the snapshot. + cp.call(t, "StatusNotification", map[string]any{"connectorId": 1, "status": "Charging", "errorCode": "NoError"}) + waitMsg(t, upRecv) // forwarded upstream + if s := sess.Snapshot(); s.ConnectorStatus != "Charging" { + t.Errorf("proxy snapshot status = %q, want Charging", s.ConnectorStatus) + } + + // Injected control: DriverVault issues RemoteStart. The charger must receive + // it and answer, and it must NOT leak to the upstream CSMS. + status, err := sess.RemoteStartTransaction(context.Background(), "TAG", 1) + if err != nil { + t.Fatalf("inject RemoteStart: %v", err) + } + if status != "Accepted" { + t.Errorf("injected RemoteStart status = %q, want Accepted", status) + } + if got := cp.waitRecv(t); got.Action != "RemoteStartTransaction" { + t.Fatalf("charger received %q, want RemoteStartTransaction", got.Action) + } + select { + case leaked := <-upRecv: + t.Fatalf("injected call leaked to upstream: %q", leaked.Action) + case <-time.After(300 * time.Millisecond): + // good — nothing forwarded upstream + } +} + +func waitMsg(t *testing.T, ch <-chan Message) Message { + t.Helper() + select { + case m := <-ch: + return m + case <-time.After(5 * time.Second): + t.Fatal("timeout awaiting message") + return Message{} + } +} diff --git a/API Server/internal/ocpp/session.go b/API Server/internal/ocpp/session.go new file mode 100644 index 0000000..cb5a4fc --- /dev/null +++ b/API Server/internal/ocpp/session.go @@ -0,0 +1,370 @@ +package ocpp + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" +) + +// Control modes. These mirror the cascade field the operator picks in Settings +// (see internal/api/integrations_ankersolix.go). +const ( + ModeOff = "off" // monitoring only; no CSMS (charger keeps its normal backend) + ModeOwn = "own" // DriverVault is the charger's Central System + ModeProxy = "proxy" // DriverVault forwards to Anker's cloud and taps/injects +) + +// injectedPrefix namespaces the message ids of CALLs DriverVault injects toward +// the charger, so that in proxy mode a reply to one of our calls is never +// confused with a reply destined for the upstream CSMS. +const injectedPrefix = "dv-" + +// callTimeout bounds how long a control command waits for the charger to answer. +const callTimeout = 30 * time.Second + +// Status is a point-in-time snapshot of one charger's OCPP state, safe to return +// to clients. +type Status struct { + Serial string `json:"serial"` + Mode string `json:"mode"` + Connected bool `json:"connected"` + Vendor string `json:"vendor,omitempty"` + Model string `json:"model,omitempty"` + Firmware string `json:"firmware,omitempty"` + BootedAt time.Time `json:"bootedAt,omitempty"` + LastHeartbeat time.Time `json:"lastHeartbeat,omitempty"` + ConnectorStatus string `json:"connectorStatus,omitempty"` // Available, Charging, Faulted, … + ErrorCode string `json:"errorCode,omitempty"` + MeterWh int64 `json:"meterWh,omitempty"` + TransactionID int `json:"transactionId,omitempty"` + LastUpdated time.Time `json:"lastUpdated,omitempty"` +} + +// CallError is returned by Session.Call when the charger answers with a CALLERROR. +type CallError struct { + Code string + Description string +} + +func (e *CallError) Error() string { + return fmt.Sprintf("ocpp charger rejected call: %s: %s", e.Code, e.Description) +} + +// callHandler answers an inbound CALL from the charger (own mode only). It +// returns either a result payload, or a non-empty errCode/errDesc to send a +// CALLERROR. +type callHandler func(s *Session, action string, payload json.RawMessage) (result any, errCode, errDesc string) + +// Session is one live charge-point connection. In own mode it also holds the +// local dispatch handler; in proxy mode it additionally holds the upstream +// connection and pumps frames between the two while tapping the stream. +type Session struct { + serial string + mode string + cp *Conn // charge-point (downstream) connection + up *Conn // upstream CSMS connection (proxy mode only) + + handler callHandler + onClose func(*Session) + logf func(string, ...any) + + mu sync.Mutex + pending map[string]chan Message // our injected call id -> result channel + status Status + closed bool + done chan struct{} + + nextTxn int // own mode: assigns transaction ids +} + +// newSession builds a session. Call start once it is registered to begin its +// read loop(s). In proxy mode up must be non-nil; in own mode handler must be +// non-nil. +func newSession(serial, mode string, cp, up *Conn, handler callHandler, onClose func(*Session), logf func(string, ...any)) *Session { + if logf == nil { + logf = func(string, ...any) {} + } + return &Session{ + serial: serial, + mode: mode, + cp: cp, + up: up, + handler: handler, + onClose: onClose, + logf: logf, + pending: map[string]chan Message{}, + done: make(chan struct{}), + status: Status{Serial: serial, Mode: mode, Connected: true}, + } +} + +// start launches the read loop(s). Separated from newSession so the CSMS can +// register the session before any inbound frame (or an immediate disconnect) can +// fire the onClose deregister callback. +func (s *Session) start() { + go s.cpLoop() + if s.mode == ModeProxy && s.up != nil { + go s.upLoop() + } +} + +// Serial returns the charger serial this session serves. +func (s *Session) Serial() string { return s.serial } + +// Mode returns the control mode (own|proxy). +func (s *Session) Mode() string { return s.mode } + +// Snapshot returns the current status. +func (s *Session) Snapshot() Status { + s.mu.Lock() + defer s.mu.Unlock() + st := s.status + st.Serial = s.serial + st.Mode = s.mode + return st +} + +// Call injects a CALL toward the charger and waits for its reply. It is how every +// control command (commands.go) reaches the charger, in both own and proxy modes. +func (s *Session) Call(ctx context.Context, action string, payload any) (json.RawMessage, error) { + ctx, cancel := context.WithTimeout(ctx, callTimeout) + defer cancel() + + id := injectedPrefix + newMessageID() + frame, err := EncodeCall(id, action, payload) + if err != nil { + return nil, err + } + ch := make(chan Message, 1) + + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil, ErrClosed + } + s.pending[id] = ch + s.mu.Unlock() + defer func() { + s.mu.Lock() + delete(s.pending, id) + s.mu.Unlock() + }() + + if err := s.cp.WriteMessage(frame); err != nil { + return nil, err + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-s.done: + return nil, ErrClosed + case m := <-ch: + if m.Type == MessageTypeCallError { + return nil, &CallError{Code: m.ErrorCode, Description: m.ErrorDescription} + } + return m.Payload, nil + } +} + +// ---- read loops -------------------------------------------------------------- + +func (s *Session) cpLoop() { + defer s.close(nil) + for { + data, err := s.cp.ReadMessage() + if err != nil { + s.close(err) + return + } + msg, err := DecodeMessage(data) + if err != nil { + s.logf("ocpp: bad frame from charger %s: %v", s.serial, err) + continue + } + s.tap(msg) + + if s.mode == ModeOwn { + s.handleLocal(msg) + continue + } + // Proxy: a reply to one of OUR injected calls is consumed locally and not + // forwarded upstream; everything else is relayed to the upstream CSMS. + if msg.Type != MessageTypeCall && s.isPending(msg.ID) { + s.deliver(msg) + continue + } + if s.up == nil || s.up.WriteMessage(data) != nil { + s.close(fmt.Errorf("ocpp: upstream write failed for %s", s.serial)) + return + } + } +} + +func (s *Session) upLoop() { + for { + data, err := s.up.ReadMessage() + if err != nil { + s.close(err) + return + } + if msg, derr := DecodeMessage(data); derr == nil { + s.tap(msg) + } + if err := s.cp.WriteMessage(data); err != nil { + s.close(err) + return + } + } +} + +// handleLocal answers an inbound frame in own mode. +func (s *Session) handleLocal(msg Message) { + switch msg.Type { + case MessageTypeCall: + result, code, desc := s.handler(s, msg.Action, msg.Payload) + var ( + out []byte + err error + ) + if code != "" { + out, err = EncodeCallError(msg.ID, code, desc, nil) + } else { + out, err = EncodeCallResult(msg.ID, result) + } + if err != nil { + s.logf("ocpp: encode response for %s/%s: %v", s.serial, msg.Action, err) + return + } + _ = s.cp.WriteMessage(out) + case MessageTypeCallResult, MessageTypeCallError: + s.deliver(msg) + } +} + +func (s *Session) isPending(id string) bool { + s.mu.Lock() + _, ok := s.pending[id] + s.mu.Unlock() + return ok +} + +func (s *Session) deliver(msg Message) { + s.mu.Lock() + ch := s.pending[msg.ID] + delete(s.pending, msg.ID) + s.mu.Unlock() + if ch != nil { + ch <- msg + } +} + +// tap updates the status snapshot from charger-originated notifications, so both +// own and proxy modes keep a live view without special-casing the dispatch path. +func (s *Session) tap(msg Message) { + if msg.Type != MessageTypeCall { + return + } + switch msg.Action { + case "BootNotification": + var p struct { + Vendor string `json:"chargePointVendor"` + Model string `json:"chargePointModel"` + Firmware string `json:"firmwareVersion"` + SerialCP string `json:"chargePointSerialNumber"` + SerialBox string `json:"chargeBoxSerialNumber"` + } + _ = json.Unmarshal(msg.Payload, &p) + s.mutate(func(st *Status) { + st.Vendor, st.Model, st.Firmware = p.Vendor, p.Model, p.Firmware + st.BootedAt = time.Now() + }) + case "Heartbeat": + s.mutate(func(st *Status) { st.LastHeartbeat = time.Now() }) + case "StatusNotification": + var p struct { + Status string `json:"status"` + ErrorCode string `json:"errorCode"` + } + _ = json.Unmarshal(msg.Payload, &p) + s.mutate(func(st *Status) { + st.ConnectorStatus = p.Status + if p.ErrorCode != "" && p.ErrorCode != "NoError" { + st.ErrorCode = p.ErrorCode + } else { + st.ErrorCode = "" + } + }) + case "MeterValues": + if wh, ok := meterWh(msg.Payload); ok { + s.mutate(func(st *Status) { st.MeterWh = wh }) + } + case "StopTransaction": + s.mutate(func(st *Status) { st.TransactionID = 0 }) + } +} + +func (s *Session) mutate(f func(*Status)) { + s.mu.Lock() + f(&s.status) + s.status.LastUpdated = time.Now() + s.mu.Unlock() +} + +func (s *Session) close(err error) { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.closed = true + s.status.Connected = false + close(s.done) + s.mu.Unlock() + + _ = s.cp.Close() + if s.up != nil { + _ = s.up.Close() + } + if s.onClose != nil { + s.onClose(s) + } + if err != nil && err != ErrClosed { + s.logf("ocpp: session %s closed: %v", s.serial, err) + } +} + +// meterWh best-effort extracts an Energy.Active.Import.Register reading (in Wh) +// from a MeterValues payload. +func meterWh(payload json.RawMessage) (int64, bool) { + var p struct { + MeterValue []struct { + SampledValue []struct { + Value string `json:"value"` + Measurand string `json:"measurand"` + Unit string `json:"unit"` + } `json:"sampledValue"` + } `json:"meterValue"` + } + if err := json.Unmarshal(payload, &p); err != nil { + return 0, false + } + for _, mv := range p.MeterValue { + for _, sv := range mv.SampledValue { + // Default measurand per spec is Energy.Active.Import.Register. + if sv.Measurand != "" && sv.Measurand != "Energy.Active.Import.Register" { + continue + } + var f float64 + if _, err := fmt.Sscanf(sv.Value, "%g", &f); err != nil { + continue + } + if sv.Unit == "kWh" { + f *= 1000 + } + return int64(f), true + } + } + return 0, false +} diff --git a/API Server/internal/ocpp/websocket.go b/API Server/internal/ocpp/websocket.go new file mode 100644 index 0000000..c39c21d --- /dev/null +++ b/API Server/internal/ocpp/websocket.go @@ -0,0 +1,438 @@ +// Package ocpp implements the DriverVault OCPP 1.6J Central System (CSMS) that +// lets the server control an EV charger — start/stop a session, cap the current, +// change availability, reset, and so on — over the persistent WebSocket the +// charger dials out to. It is a separate long-lived subsystem from the read-only +// Anker Solix cloud plugin (internal/plugins/builtin/ankersolix), which can only +// monitor. +// +// The whole API server is dependency-free (stdlib only), so the WebSocket layer +// here is a hand-rolled RFC 6455 implementation rather than gorilla/websocket. It +// is deliberately minimal: text-message oriented (OCPP frames are JSON text), +// single reader / serialized writer, with ping/pong and close handled inline. +package ocpp + +import ( + "bufio" + "context" + "crypto/rand" + "crypto/sha1" + "crypto/tls" + "encoding/base64" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "sync" + "time" +) + +// wsGUID is the RFC 6455 magic value appended to Sec-WebSocket-Key when deriving +// the Sec-WebSocket-Accept response. +const wsGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +// maxMessageBytes caps a single assembled message; OCPP payloads are small, so +// this is a generous ceiling that also bounds a hostile peer. +const maxMessageBytes = 1 << 20 // 1 MiB + +// WebSocket opcodes (RFC 6455 §5.2). +const ( + opContinuation = 0x0 + opText = 0x1 + opBinary = 0x2 + opClose = 0x8 + opPing = 0x9 + opPong = 0xA +) + +// ErrClosed is returned by ReadMessage once the peer has sent a close frame. +var ErrClosed = errors.New("ocpp: connection closed") + +// Conn is a minimal RFC 6455 connection, usable as either the server (CSMS) or +// client (proxy→upstream, and the test charge point) end. The only difference is +// masking: per the spec a client masks the frames it sends, a server does not. +type Conn struct { + raw net.Conn + br *bufio.Reader + isServer bool + + wmu sync.Mutex // serializes all writes (data + control frames) + + closeOnce sync.Once + closed chan struct{} +} + +// Upgrade performs the server-side handshake on a hijackable ResponseWriter and +// returns a ready Conn. On success the caller owns the connection and must not +// touch w again. It negotiates the "ocpp1.6" subprotocol when offered. +func Upgrade(w http.ResponseWriter, r *http.Request) (*Conn, error) { + if !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") || + !headerHasToken(r.Header.Get("Connection"), "upgrade") { + return nil, errors.New("ocpp: not a websocket upgrade request") + } + key := r.Header.Get("Sec-WebSocket-Key") + if key == "" { + return nil, errors.New("ocpp: missing Sec-WebSocket-Key") + } + hj, ok := w.(http.Hijacker) + if !ok { + return nil, errors.New("ocpp: response writer does not support hijacking") + } + conn, brw, err := hj.Hijack() + if err != nil { + return nil, err + } + + var subproto string + for _, p := range splitTokens(r.Header.Get("Sec-WebSocket-Protocol")) { + if strings.EqualFold(p, "ocpp1.6") { + subproto = "ocpp1.6" + break + } + } + + var b strings.Builder + b.WriteString("HTTP/1.1 101 Switching Protocols\r\n") + b.WriteString("Upgrade: websocket\r\n") + b.WriteString("Connection: Upgrade\r\n") + b.WriteString("Sec-WebSocket-Accept: " + acceptKey(key) + "\r\n") + if subproto != "" { + b.WriteString("Sec-WebSocket-Protocol: " + subproto + "\r\n") + } + b.WriteString("\r\n") + if _, err := conn.Write([]byte(b.String())); err != nil { + _ = conn.Close() + return nil, err + } + return &Conn{raw: conn, br: brw.Reader, isServer: true, closed: make(chan struct{})}, nil +} + +// Dial opens a client WebSocket to rawURL (ws:// or wss://), offering the given +// subprotocols and any extra request headers (e.g. Authorization for OCPP +// security profile 1). It is used for proxy→upstream links and by tests. +func Dial(ctx context.Context, rawURL string, subprotocols []string, header http.Header) (*Conn, error) { + u, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("ocpp: bad url: %w", err) + } + var ( + secure bool + port = u.Port() + ) + switch strings.ToLower(u.Scheme) { + case "ws", "http": + if port == "" { + port = "80" + } + case "wss", "https": + secure, port = true, orDefault(port, "443") + default: + return nil, fmt.Errorf("ocpp: unsupported scheme %q", u.Scheme) + } + + d := &net.Dialer{} + raw, err := d.DialContext(ctx, "tcp", net.JoinHostPort(u.Hostname(), port)) + if err != nil { + return nil, err + } + if secure { + tconn := tls.Client(raw, &tls.Config{ServerName: u.Hostname()}) + if err := tconn.HandshakeContext(ctx); err != nil { + _ = raw.Close() + return nil, err + } + raw = tconn + } + if dl, ok := ctx.Deadline(); ok { + _ = raw.SetDeadline(dl) + } + + var keyBytes [16]byte + if _, err := rand.Read(keyBytes[:]); err != nil { + _ = raw.Close() + return nil, err + } + key := base64.StdEncoding.EncodeToString(keyBytes[:]) + + reqPath := u.RequestURI() + if reqPath == "" { + reqPath = "/" + } + var b strings.Builder + b.WriteString("GET " + reqPath + " HTTP/1.1\r\n") + b.WriteString("Host: " + u.Host + "\r\n") + b.WriteString("Upgrade: websocket\r\n") + b.WriteString("Connection: Upgrade\r\n") + b.WriteString("Sec-WebSocket-Version: 13\r\n") + b.WriteString("Sec-WebSocket-Key: " + key + "\r\n") + if len(subprotocols) > 0 { + b.WriteString("Sec-WebSocket-Protocol: " + strings.Join(subprotocols, ", ") + "\r\n") + } + for k, vs := range header { + for _, v := range vs { + b.WriteString(k + ": " + v + "\r\n") + } + } + b.WriteString("\r\n") + if _, err := raw.Write([]byte(b.String())); err != nil { + _ = raw.Close() + return nil, err + } + + br := bufio.NewReader(raw) + resp, err := http.ReadResponse(br, &http.Request{Method: http.MethodGet}) + if err != nil { + _ = raw.Close() + return nil, err + } + resp.Body.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + _ = raw.Close() + return nil, fmt.Errorf("ocpp: upstream refused upgrade (HTTP %d)", resp.StatusCode) + } + if got := resp.Header.Get("Sec-WebSocket-Accept"); got != acceptKey(key) { + _ = raw.Close() + return nil, errors.New("ocpp: bad Sec-WebSocket-Accept from upstream") + } + // Clear the dial deadline; per-operation deadlines are set explicitly later. + _ = raw.SetDeadline(time.Time{}) + return &Conn{raw: raw, br: br, isServer: false, closed: make(chan struct{})}, nil +} + +// ReadMessage returns the next complete text/binary message, transparently +// answering ping frames and honoring a close frame (after which it returns +// ErrClosed). Control frames never surface to the caller. +func (c *Conn) ReadMessage() ([]byte, error) { + var ( + buf []byte + started bool + ) + for { + f, err := c.readFrame() + if err != nil { + return nil, err + } + switch f.opcode { + case opPing: + _ = c.writeFrame(opPong, f.data) + continue + case opPong: + continue + case opClose: + _ = c.writeFrame(opClose, f.data) + c.markClosed() + return nil, ErrClosed + case opText, opBinary: + if started { + return nil, errors.New("ocpp: new data frame before previous finished") + } + started = true + buf = append(buf, f.data...) + case opContinuation: + if !started { + return nil, errors.New("ocpp: continuation frame with no start") + } + buf = append(buf, f.data...) + default: + return nil, fmt.Errorf("ocpp: unexpected opcode 0x%x", f.opcode) + } + if len(buf) > maxMessageBytes { + return nil, errors.New("ocpp: message too large") + } + if f.fin { + return buf, nil + } + } +} + +// WriteMessage sends one OCPP text frame. +func (c *Conn) WriteMessage(data []byte) error { + select { + case <-c.closed: + return ErrClosed + default: + } + return c.writeFrame(opText, data) +} + +// SetReadDeadline bounds the next read (used to detect a dead peer between +// heartbeats). A zero time clears it. +func (c *Conn) SetReadDeadline(t time.Time) error { return c.raw.SetReadDeadline(t) } + +// Close sends a close frame (best effort) and shuts the underlying connection. +func (c *Conn) Close() error { + var err error + c.closeOnce.Do(func() { + _ = c.writeFrame(opClose, nil) + close(c.closed) + err = c.raw.Close() + }) + return err +} + +// markClosed records that the peer initiated close without double-sending. +func (c *Conn) markClosed() { + c.closeOnce.Do(func() { + close(c.closed) + _ = c.raw.Close() + }) +} + +// ---- framing ----------------------------------------------------------------- + +type frame struct { + fin bool + opcode byte + data []byte +} + +func (c *Conn) readFrame() (frame, error) { + var h [2]byte + if _, err := io.ReadFull(c.br, h[:]); err != nil { + return frame{}, err + } + fin := h[0]&0x80 != 0 + if h[0]&0x70 != 0 { + return frame{}, errors.New("ocpp: reserved bits set (no extensions supported)") + } + opcode := h[0] & 0x0f + masked := h[1]&0x80 != 0 + length := uint64(h[1] & 0x7f) + switch length { + case 126: + var ext [2]byte + if _, err := io.ReadFull(c.br, ext[:]); err != nil { + return frame{}, err + } + length = uint64(binary.BigEndian.Uint16(ext[:])) + case 127: + var ext [8]byte + if _, err := io.ReadFull(c.br, ext[:]); err != nil { + return frame{}, err + } + length = binary.BigEndian.Uint64(ext[:]) + } + if length > maxMessageBytes { + return frame{}, errors.New("ocpp: frame too large") + } + // RFC 6455: the server must receive masked frames; a client must receive + // unmasked ones. + if c.isServer && !masked { + return frame{}, errors.New("ocpp: unmasked frame from client") + } + if !c.isServer && masked { + return frame{}, errors.New("ocpp: masked frame from server") + } + var maskKey [4]byte + if masked { + if _, err := io.ReadFull(c.br, maskKey[:]); err != nil { + return frame{}, err + } + } + data := make([]byte, length) + if _, err := io.ReadFull(c.br, data); err != nil { + return frame{}, err + } + if masked { + for i := range data { + data[i] ^= maskKey[i%4] + } + } + return frame{fin: fin, opcode: opcode, data: data}, nil +} + +func (c *Conn) writeFrame(opcode byte, data []byte) error { + c.wmu.Lock() + defer c.wmu.Unlock() + + mask := !c.isServer // clients mask their frames + var head [14]byte + head[0] = 0x80 | opcode // FIN + opcode (we never fragment outgoing messages) + n := 2 + l := len(data) + switch { + case l <= 125: + head[1] = byte(l) + case l <= 0xFFFF: + head[1] = 126 + binary.BigEndian.PutUint16(head[2:4], uint16(l)) + n = 4 + default: + head[1] = 127 + binary.BigEndian.PutUint64(head[2:10], uint64(l)) + n = 10 + } + + var maskKey [4]byte + if mask { + head[1] |= 0x80 + if _, err := rand.Read(maskKey[:]); err != nil { + return err + } + copy(head[n:n+4], maskKey[:]) + n += 4 + } + + _ = c.raw.SetWriteDeadline(time.Now().Add(30 * time.Second)) + defer c.raw.SetWriteDeadline(time.Time{}) + + if _, err := c.raw.Write(head[:n]); err != nil { + return err + } + if l == 0 { + return nil + } + if mask { + masked := make([]byte, l) + for i := 0; i < l; i++ { + masked[i] = data[i] ^ maskKey[i%4] + } + _, err := c.raw.Write(masked) + return err + } + _, err := c.raw.Write(data) + return err +} + +// ---- small helpers ----------------------------------------------------------- + +func acceptKey(key string) string { + h := sha1.New() + _, _ = io.WriteString(h, key+wsGUID) + return base64.StdEncoding.EncodeToString(h.Sum(nil)) +} + +// headerHasToken reports whether a comma-separated header value contains token +// (case-insensitive) — e.g. Connection: keep-alive, Upgrade. +func headerHasToken(value, token string) bool { + for _, t := range splitTokens(value) { + if strings.EqualFold(t, token) { + return true + } + } + return false +} + +func splitTokens(value string) []string { + if value == "" { + return nil + } + parts := strings.Split(value, ",") + out := make([]string, 0, len(parts)) + for _, p := range parts { + if p = strings.TrimSpace(p); p != "" { + out = append(out, p) + } + } + return out +} + +func orDefault(v, def string) string { + if v == "" { + return def + } + return v +} diff --git a/API Server/internal/ocpp/websocket_test.go b/API Server/internal/ocpp/websocket_test.go new file mode 100644 index 0000000..4304565 --- /dev/null +++ b/API Server/internal/ocpp/websocket_test.go @@ -0,0 +1,79 @@ +package ocpp + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestAcceptKey(t *testing.T) { + // RFC 6455 §1.3 worked example. + if got := acceptKey("dGhlIHNhbXBsZSBub25jZQ=="); got != "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" { + t.Fatalf("acceptKey = %q, want s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", got) + } +} + +// echoServer upgrades and echoes every message back, exercising the framing + +// masking round trip in both directions across a range of payload sizes. +func TestFramingRoundTrip(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := Upgrade(w, r) + if err != nil { + t.Errorf("upgrade: %v", err) + return + } + for { + msg, err := conn.ReadMessage() + if err != nil { + return + } + if err := conn.WriteMessage(msg); err != nil { + return + } + } + })) + defer srv.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + conn, err := Dial(ctx, wsURL(srv.URL, "/"), []string{"ocpp1.6"}, nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + + // Small (<126), medium (2-byte length), large (8-byte length) — all masked + // client→server, unmasked on the way back. + for _, size := range []int{5, 200, 70000} { + want := bytes.Repeat([]byte("x"), size) + if err := conn.WriteMessage(want); err != nil { + t.Fatalf("write size %d: %v", size, err) + } + got, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read size %d: %v", size, err) + } + if !bytes.Equal(got, want) { + t.Fatalf("size %d: echoed %d bytes, want %d", size, len(got), len(want)) + } + } +} + +func TestUpgradeRejectsNonWebSocket(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/ocpp/CP1", nil) + if _, err := Upgrade(rec, req); err == nil { + t.Fatal("expected error upgrading a plain GET") + } +} + +func TestDialRejectsBadScheme(t *testing.T) { + if _, err := Dial(context.Background(), "ftp://example/x", nil, nil); err == nil || + !strings.Contains(err.Error(), "unsupported scheme") { + t.Fatalf("want unsupported scheme error, got %v", err) + } +} diff --git a/API Server/internal/plugins/builtin/ankersolix/ankersolix.go b/API Server/internal/plugins/builtin/ankersolix/ankersolix.go index 4d0acf2..5a6afe0 100644 --- a/API Server/internal/plugins/builtin/ankersolix/ankersolix.go +++ b/API Server/internal/plugins/builtin/ankersolix/ankersolix.go @@ -143,6 +143,17 @@ func (p *Plugin) Descriptor() plugins.Descriptor { Help: "Your Anker account password. Stored locally, sent only (encrypted) to Anker's login endpoint."}, {Key: "country", Label: "Country", Type: "text", Default: "DE", Help: "ISO country code of your Anker account (e.g. DE, GB, US). Determines which Anker server the account lives on; a wrong value logs in but shows no devices."}, + // controlMode selects the OCPP control path. It does not affect this + // (read-only) cloud plugin — the CSMS that acts on it lives in the api + // package (internal/ocpp) — but it is advertised here so a superadmin can + // set/lock it at the global layer, and it cascades like the other fields. + {Key: "controlMode", Label: "Control mode", Type: "select", Default: "off", + Help: "How DriverVault controls the charger over OCPP. Off = monitoring only (default). Own CSMS = the charger connects directly to DriverVault. Proxy CSMS = DriverVault relays to Anker's cloud and can inject commands.", + Options: []plugins.SelectOption{ + {Value: "off", Label: "Off (monitoring only)"}, + {Value: "own", Label: "Own CSMS (full control)"}, + {Value: "proxy", Label: "Proxy CSMS (relay + control)"}, + }}, }, } } diff --git a/API Server/internal/plugins/builtin/ankersolix/ankersolix_test.go b/API Server/internal/plugins/builtin/ankersolix/ankersolix_test.go index 6eda07b..59bc582 100644 --- a/API Server/internal/plugins/builtin/ankersolix/ankersolix_test.go +++ b/API Server/internal/plugins/builtin/ankersolix/ankersolix_test.go @@ -43,6 +43,29 @@ func TestDescriptor(t *testing.T) { if !fields["password"].Secret { t.Error("password field must be marked secret") } + + // controlMode is a select advertising the three OCPP control paths. + cm, ok := fields["controlMode"] + if !ok { + t.Fatal("controlMode config field should be present") + } + if cm.Type != "select" { + t.Errorf("controlMode type = %q, want select", cm.Type) + } + if cm.Default != "off" { + t.Errorf("controlMode default = %q, want off", cm.Default) + } + want := map[string]bool{"off": false, "own": false, "proxy": false} + for _, o := range cm.Options { + if _, known := want[o.Value]; known { + want[o.Value] = true + } + } + for v, seen := range want { + if !seen { + t.Errorf("controlMode is missing option %q", v) + } + } } func TestRegistered(t *testing.T) { diff --git a/Web App/web/src/api.js b/Web App/web/src/api.js index e811205..1684976 100644 --- a/Web App/web/src/api.js +++ b/Web App/web/src/api.js @@ -239,6 +239,20 @@ export const api = { saveAnkerSolix: (body) => request("/integrations/anker-solix", { method: "PUT", body: JSON.stringify(body) }), testAnkerSolix: () => request("/integrations/anker-solix/health", { method: "POST" }), + // Anker Solix OCPP control (per charger). getAnkerControl returns the control + // mode, connection status, provisioning endpoint + token, and a live status + // snapshot; ankerControlToken (re)generates the per-charger token the operator + // installs into the charger; ankerControlAction issues one OCPP command + // (start/stop/limit/clear-limit/availability/reset/unlock/trigger/config). + getAnkerControl: (sn) => request(`/integrations/anker-solix/chargers/${encodeURIComponent(sn)}/control`), + ankerControlToken: (sn) => + request(`/integrations/anker-solix/chargers/${encodeURIComponent(sn)}/control/token`, { method: "POST" }), + ankerControlAction: (sn, action, body = {}) => + request(`/integrations/anker-solix/chargers/${encodeURIComponent(sn)}/${action}`, { + method: "POST", + body: JSON.stringify(body), + }), + // Settings — advanced / danger zone exportData: () => requestBlob("/me/export"), importData: (payload) => request("/me/import", { method: "POST", body: JSON.stringify(payload) }), diff --git a/Web App/web/src/i18n/en.json b/Web App/web/src/i18n/en.json index 3c7206d..e7be865 100644 --- a/Web App/web/src/i18n/en.json +++ b/Web App/web/src/i18n/en.json @@ -47,6 +47,22 @@ "idle": "Not charging", "idleHint": "Plug in at a station to start a session." }, + "control": { + "title": "Charger control", + "connected": "Connected", + "disconnected": "Offline", + "serialPlaceholder": "Charger serial (e.g. A5191-XXXXXXXX)", + "refresh": "Refresh", + "status": "Connector", + "meter": "Energy", + "start": "Start charging", + "stop": "Stop charging", + "limit": "Current limit", + "applyLimit": "Apply limit", + "clearLimit": "Clear limit", + "reset": "Reset charger", + "connectHint": "Enter your charger's serial and refresh. The charger must be connected to DriverVault's OCPP backend (set up in Settings → Integrations)." + }, "stations": { "heading": "Nearby", "count": { @@ -222,7 +238,23 @@ "ankerEmail": "Anker account email", "ankerPassword": "Anker account password", "country": "Country", - "countryHint": "Two-letter country code of your Anker account (e.g. DE, GB, US)." + "countryHint": "Two-letter country code of your Anker account (e.g. DE, GB, US).", + "controlMode": "Control mode", + "controlModeHint": "How DriverVault controls the charger over OCPP.", + "controlOff": "Off (monitoring only)", + "controlOwn": "Own CSMS (full control)", + "controlProxy": "Proxy CSMS (relay + control)", + "controlTitle": "Charger control (OCPP)", + "controlOwnHint": "The charger connects directly to DriverVault as its Central System. Point the charger's OCPP backend at the endpoint below.", + "controlProxyHint": "DriverVault relays to Anker's cloud and can inject commands. Point the charger's OCPP backend at the endpoint below.", + "controlProvisionSteps": "In the Anker app (or the charger's OCPP settings), set the OCPP backend URL to the endpoint and the authorization key to the token above.", + "chargerSerial": "Charger serial", + "controlCheck": "Check status", + "controlGenerate": "Generate token", + "controlEndpoint": "OCPP endpoint", + "controlToken": "Auth token", + "controlConnected": "Connected to control backend", + "controlDisconnected": "Not connected" }, "privacy": { diff --git a/Web App/web/src/views/Charging.vue b/Web App/web/src/views/Charging.vue index b6fd425..e86d3e5 100644 --- a/Web App/web/src/views/Charging.vue +++ b/Web App/web/src/views/Charging.vue @@ -1,6 +1,7 @@