Merge pull request #11 from smlx/batman

feat: add batman metrics
smlx · Dec 1, 2023 · f3f361b · f3f361b
2 parents 3515e65 + 6e41a0a
commit f3f361b
Show file tree

Hide file tree

Showing 11 changed files with 795 additions and 359 deletions.
diff --git a/cmd/sems_mitm_exporter/serve.go b/cmd/sems_mitm_exporter/serve.go
@@ -19,10 +19,12 @@ const (
 )
 
 // ServeCmd represents the `serve` command.
-type ServeCmd struct{}
+type ServeCmd struct {
+	Batsignal bool `kong:"env='BATSIGNAL',help='Enable Batsignal mode (draws the bat-insignia on the SEMS portal graph)'"`
+}
 
 // Run the serve command.
-func (*ServeCmd) Run(log *slog.Logger) error {
+func (cmd *ServeCmd) Run(log *slog.Logger) error {
 	// handle signals
 	ctx, stop := signal.NotifyContext(
 		context.Background(),
@@ -57,7 +59,7 @@ func (*ServeCmd) Run(log *slog.Logger) error {
 	})
 	// start mitm server
 	eg.Go(func() error {
-		return mitm.Serve(ctx, log)
+		return mitm.NewServer(cmd.Batsignal).Serve(ctx, log)
 	})
 	return eg.Wait()
 }
diff --git a/mitm/batsignal.go b/mitm/batsignal.go
@@ -0,0 +1,90 @@
+package mitm
+
+import (
+	"math"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type timeFunc func() time.Time
+
+func timeNow() time.Time {
+	return time.Now().In(time.FixedZone("+08", 8*60*60))
+}
+
+// setupBatsignal enables the prometheus metrics which display the bat insignia
+// on the time series graph
+func setupBatsignal() {
+	promauto.NewUntypedFunc(prometheus.UntypedOpts{
+		Name: "batsignal_top",
+		Help: "Top of the batsignal",
+	}, batsignalTop(timeNow))
+	promauto.NewUntypedFunc(prometheus.UntypedOpts{
+		Name: "batsignal_bottom",
+		Help: "Bottom of the batsignal",
+	}, batsignalBottom(timeNow))
+}
+
+// Taking the time returned by tf as an x value zeroed to midday, calculates
+// the positive y value of the Batman function.
+// https://www.pacifict.com/Examples/Batman/
+func batsignalTop(tf timeFunc) func() float64 {
+	return func() float64 {
+		now := tf()
+		switch x := float64(now.Hour()) + float64(now.Minute())/60 - 12; {
+		case x > -7 && x < -3:
+			fallthrough
+		case x > 3 && x < 7:
+			return math.Sqrt(1-math.Pow(x/7, 2)) * 3
+		case x >= -3 && x < -1:
+			fallthrough
+		case x > 1 && x <= 3:
+			return 6*math.Sqrt(10)/7 - 0.5*math.Abs(x) + 1.5 -
+				(3*math.Sqrt(10)/7)*math.Sqrt(4-math.Pow(math.Abs(x)-1, 2))
+		case x >= -1 && x < -0.75:
+			fallthrough
+		case x > 0.75 && x <= 1:
+			return 9 - 8*math.Abs(x)
+		case x >= -0.75 && x < -0.5:
+			fallthrough
+		case x > 0.5 && x <= 0.75:
+			return 3*math.Abs(x) + 0.75
+		case x >= -0.5 && x <= 0.5:
+			return 2.25
+		default:
+			return 0
+		}
+	}
+}
+
+// Taking the time returned by tf as an x value zeroed to midday, calculates
+// the negative y value of the Batman function.
+// https://www.pacifict.com/Examples/Batman/
+func batsignalBottom(tf timeFunc) func() float64 {
+	return func() float64 {
+		now := tf()
+		switch x := float64(now.Hour()) + float64(now.Minute())/60 - 12; {
+		case x > -7 && x < -4:
+			fallthrough
+		case x > 4 && x < 7:
+			return -math.Sqrt(1-math.Pow(x/7, 2)) * 3
+		case x >= -4 && x <= 4:
+			return math.Abs(x/2) - (3*math.Sqrt(33)-7)/112*math.Pow(x, 2) - 3 +
+				math.Sqrt(1-math.Pow(math.Abs(math.Abs(x)-2)-1, 2))
+		default:
+			return 0
+		}
+	}
+}
+
+// batsignal takes data assumed to be an outbound packet. If it is a metrics
+// packet, it mutates the metrics values of the data in order to draw the
+// Batman function on the SEMS portal plot.
+// On any kind of error, it returns the original data.
+func batsignal(metrics *OutboundMetricsPacket) ([]byte, error) {
+	metrics.PowerGenerationWatts = int32(1000 * batsignalTop(timeNow)())
+	metrics.PowerExportWatts = int32(1000 * batsignalBottom(timeNow)())
+	return metrics.MarshalBinary()
+}
diff --git a/mitm/inbound.go b/mitm/inbound.go
@@ -7,13 +7,9 @@ import (
 	"fmt"
 	"log/slog"
 	"slices"
-)
 
-const (
-	// packet structure constants
-	inboundEnvelopeLen = 0x20
-	// cleartext body constants
-	timeSyncRespLen = 0x10
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
 var (
@@ -24,150 +20,134 @@ var (
 	packetTypeMetricsAck1  = []byte{0x03, 0x45}
 	packetTypeMetricsAck2  = []byte{0x03, 0x03}
 	packetTypeTimeSyncResp = []byte{0x01, 0x16}
+	// protocol constants
+	// metricsAckData is sent by the server when it receives data sucessfully
+	metricsAckData = []byte{
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+	}
+	// metricsNackData is sent by the server when it receives data unsucessfully (e.g. bad CRC)
+	metricsNackData = []byte{
+		0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+	}
+	// prometheus metrics
+	inboundUnknownPacketsTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "inbound_unknown_packets_total",
+		Help: "Count of outbound unknown packets.",
+	})
 )
 
-// InboundHeader represents the header of an inbound packet.
-type InboundHeader struct {
-	GW         [2]byte // GW prefix
-	Length     uint32  // Off-by-one
-	PacketType [2]byte // Packet type identifier
-}
-
-// InboundEnvelope is the plaintext wrapper around the ciphertext.
-type InboundEnvelope struct {
-	DeviceID     [8]byte  // ASCII
-	DeviceSerial [8]byte  // ASCII
-	IV           [16]byte // AES-128 Initialization Vector
-}
-
-// InboundTimeSyncResp is the cleartext body of an inbound time sync packet.
-type InboundTimeSyncResp struct {
-	PacketType   [4]byte   // 0x00-0x03 Packet type?
-	Timestamp    Timestamp // 0x04-0x09 Y M D H m s
-	UnknownBytes [6]byte   // 0x0a-0x0f Fixed null?
-}
-
 // handleMetricsAckPacket handles metrics ack packet envelope and ciphertext.
 func handleMetricsAckPacket(
-	buf *bytes.Buffer,
-	headerLen uint32,
+	data []byte,
 	log *slog.Logger,
 ) error {
-	envelope := InboundEnvelope{}
-	if err := binary.Read(buf, binary.BigEndian, &envelope); err != nil {
-		return fmt.Errorf("couldn't read envelope: %v", err)
-	}
-	// -2 for packet type field in header and +1 for off-by-one = -1
-	ciphertext := buf.Next(int(headerLen - inboundEnvelopeLen - 1))
-	cleartext, err := decryptCiphertext(envelope.IV[:], ciphertext)
+	var inboundMetricsAck InboundMetricsAckPacket
+	err := inboundMetricsAck.UnmarshalBinary(data)
 	if err != nil {
-		log.Debug("couldn't decrypt ciphertext", slog.Any("ciphertext", ciphertext))
-		return fmt.Errorf("couldn't decrypt ciphertext: %v", err)
+		return fmt.Errorf("couldn't unmarshal metrics ack: %v", err)
 	}
-	if !slices.Equal(metricsAck, cleartext) {
-		log.Debug("unknown cleartext in metrics ack",
-			slog.Any("cleartext", cleartext))
-		return fmt.Errorf("unknown cleartext in metrics ack")
+	switch {
+	case slices.Equal(inboundMetricsAck.Data[:], metricsAckData):
+		log.Debug("metrics ack")
+	case slices.Equal(inboundMetricsAck.Data[:], metricsNackData):
+		log.Warn("metrics nack. bad metrics CRC?")
+	default:
+		log.Warn("unknown cleartext in metrics ack",
+			slog.Any("cleartext", inboundMetricsAck.Data[:]))
 	}
-	log.Debug("metrics ack")
 	return nil
 }
 
-// parseTimeSyncResp unmarshals the time sync response body.
-func parseTimeSyncResp(cleartext []byte) (*InboundTimeSyncResp, error) {
-	if len(cleartext) != timeSyncRespLen {
-		return nil, fmt.Errorf("invalid cleartext length: %d", len(cleartext))
-	}
-	body := InboundTimeSyncResp{}
-	buf := bytes.NewBuffer(cleartext)
-	if err := binary.Read(buf, binary.BigEndian, &body); err != nil {
-		return nil, fmt.Errorf("couldn't read cleartext: %v", err)
-	}
-	return &body, nil
-}
-
 // handleTimeSyncRespPacket handles time sync response packet envelope and
 // ciphertext.
 func handleTimeSyncRespPacket(
-	buf *bytes.Buffer,
-	headerLen uint32,
+	data []byte,
 	log *slog.Logger,
 ) error {
-	envelope := InboundEnvelope{}
-	if err := binary.Read(buf, binary.BigEndian, &envelope); err != nil {
-		return fmt.Errorf("couldn't read envelope: %v", err)
-	}
-	// -2 for packet type field in header and +1 for off-by-one = -1
-	ciphertext := buf.Next(int(headerLen - inboundEnvelopeLen - 1))
-	cleartext, err := decryptCiphertext(envelope.IV[:], ciphertext)
-	if err != nil {
-		log.Debug("couldn't decrypt ciphertext", slog.Any("ciphertext", ciphertext))
-		return fmt.Errorf("couldn't decrypt ciphertext: %v", err)
-	}
-	tsResp, err := parseTimeSyncResp(cleartext)
+	var timeSyncResp InboundTimeSyncRespPacket
+	err := timeSyncResp.UnmarshalBinary(data)
 	if err != nil {
-		return fmt.Errorf("couldn't parse time sync response: %v", err)
+		return fmt.Errorf("couldn't unmarshal time sync response: %v", err)
 	}
 	log.Debug("inbound time sync response",
-		slog.Time("responseTimestamp", tsResp.Timestamp.Time()))
+		slog.Time("responseTimestamp", timeSyncResp.Timestamp.Time()))
 	return nil
 }
 
 // handleUnknownInboundPacket decrypts and logs the cleartext of an
 // unrecognized inbound packet.
 func handleUnknownInboundPacket(
-	buf *bytes.Buffer,
-	headerLen uint32,
+	data []byte,
 	log *slog.Logger,
 ) error {
+	log.Info("unknown packet", slog.Any("data", data))
+	inboundUnknownPacketsTotal.Inc()
 	envelope := InboundEnvelope{}
-	if err := binary.Read(buf, binary.BigEndian, &envelope); err != nil {
-		return fmt.Errorf("couldn't read envelope: %v", err)
+	envData, bodyData := data[:binary.Size(envelope)], data[binary.Size(envelope):]
+	envBuf := bytes.NewBuffer(envData)
+	err := binary.Read(envBuf, binary.BigEndian, &envelope)
+	if err != nil {
+		return fmt.Errorf("couldn't unmarshal envelope %T: %v", envelope, err)
 	}
-	// -2 for packet type field in header and +1 for off-by-one = -1
-	ciphertext := buf.Next(int(headerLen - inboundEnvelopeLen - 1))
-	cleartext, err := decryptCiphertext(envelope.IV[:], ciphertext)
+	cleartext, err := decryptCiphertext(envelope.IV[:], bodyData)
 	if err != nil {
-		log.Debug("couldn't decrypt ciphertext", slog.Any("ciphertext", ciphertext))
 		return fmt.Errorf("couldn't decrypt ciphertext: %v", err)
 	}
-	log.Info("unknown packet", slog.Any("cleartext", cleartext))
+	log.Info("unknown packet cleartext", slog.Any("cleartext", cleartext))
 	return nil
 }
 
-// handleInboundPacket is a handlePacketFunc for inbound packets.
-func handleInboundPacket(
+// InboundPacketHandler is a PacketHandler for inbound packets.
+type InboundPacketHandler struct{}
+
+// NewInboundPacketHandler constructs an InboundPacketHandler.
+func NewInboundPacketHandler() *InboundPacketHandler {
+	return &InboundPacketHandler{}
+}
+
+// HandlePacket implements the PacketHandler interface.
+func (h *InboundPacketHandler) HandlePacket(
 	ctx context.Context,
 	log *slog.Logger,
 	data []byte,
-) error {
+) ([]byte, error) {
 	if err := validateCRC(data, inboundCRCByteOrder); err != nil {
-		return fmt.Errorf("couldn't validate CRC: %v", err)
+		return nil, fmt.Errorf("couldn't validate CRC: %v", err)
 	}
+	// slice up the header and body, and discard CRC bytes
 	header := InboundHeader{}
-	buf := bytes.NewBuffer(data)
-	if err := binary.Read(buf, binary.BigEndian, &header); err != nil {
-		return fmt.Errorf("couldn't read header: %v", err)
+	headerData, bodyData :=
+		data[:binary.Size(header)], data[binary.Size(header):len(data)-2]
+	if err := header.UnmarshalBinary(headerData); err != nil {
+		return nil, fmt.Errorf("couldn't unmarshal header: %v", err)
+	}
+	// validate data size: -2 for packet type field and +1 for length off-by-one = -1
+	expectedBodySize := header.Length - 1
+	if len(bodyData) != int(expectedBodySize) {
+		return nil, fmt.Errorf("expected body size %d, got %d",
+			expectedBodySize, len(bodyData))
 	}
 	switch {
 	case slices.Equal(packetTypeMetricsAck0, header.PacketType[:]):
 		fallthrough
 	case slices.Equal(packetTypeMetricsAck1, header.PacketType[:]):
 		fallthrough
 	case slices.Equal(packetTypeMetricsAck2, header.PacketType[:]):
-		if err := handleMetricsAckPacket(buf, header.Length, log); err != nil {
-			return fmt.Errorf("couldn't handle metrics ack packet: %v", err)
+		if err := handleMetricsAckPacket(bodyData, log); err != nil {
+			return nil, fmt.Errorf("couldn't handle metrics ack packet: %v", err)
 		}
-		return nil
+		return nil, nil
 	case slices.Equal(packetTypeTimeSyncResp, header.PacketType[:]):
-		if err := handleTimeSyncRespPacket(buf, header.Length, log); err != nil {
-			return fmt.Errorf("couldn't handle time sync response packet: %v", err)
+		if err := handleTimeSyncRespPacket(bodyData, log); err != nil {
+			return nil, fmt.Errorf("couldn't handle time sync response packet: %v", err)
 		}
-		return nil
+		return nil, nil
 	default:
-		if err := handleUnknownInboundPacket(buf, header.Length, log); err != nil {
-			return fmt.Errorf("couldn't handle unknown packet: %v", err)
+		if err := handleUnknownInboundPacket(bodyData, log); err != nil {
+			return nil, fmt.Errorf("couldn't handle unknown packet: %v", err)
 		}
-		return fmt.Errorf("unknown packet type")
+		return nil, fmt.Errorf("unknown packet type")
 	}
 }
diff --git a/mitm/inbound_test.go b/mitm/inbound_test.go
@@ -68,12 +68,14 @@ func TestHandleInboundPacket(t *testing.T) {
 	ctx := context.Background()
 	log := slog.New(slog.NewJSONHandler(os.Stderr,
 		&slog.HandlerOptions{Level: slog.LevelDebug}))
+	ph := NewInboundPacketHandler()
 	for name, tc := range testCases {
 		t.Run(name, func(tt *testing.T) {
+			_, err := ph.HandlePacket(ctx, log, tc.input)
 			if tc.expectError {
-				assert.Error(tt, handleInboundPacket(ctx, log, tc.input), name)
+				assert.Error(tt, err, name)
 			} else {
-				assert.NoError(tt, handleInboundPacket(ctx, log, tc.input), name)
+				assert.NoError(tt, err, name)
 			}
 			deviceSerial, err := deviceSerialInbound(tc.input)
 			assert.NoError(tt, err, name)
@@ -203,8 +205,9 @@ func TestHandleInbound(t *testing.T) {
 				return err
 			})
 			// test the function
-			assert.NoError(tt, handleConn(ctx, log, upstreamRead, clientWrite,
-				inboundPrefix, false, handleInboundPacket), name)
+			mitmSrv := NewServer(false)
+			assert.NoError(tt, mitmSrv.handleConn(ctx, log, upstreamRead, clientWrite,
+				inboundPrefix, false, NewInboundPacketHandler()), name)
 			if err := eg.Wait(); err != nil {
 				tt.Fatal(err)
 			}