cloudinit-bot
29c5947fa5
---ci--- project: oy phase: 2 milestone: v0.5 status: complete requirements: covered: [REQ-034] partial: [] ---/ci---
388 lines
14 KiB
Go
388 lines
14 KiB
Go
package keeper
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
sdk "github.com/cosmos/cosmos-sdk/types"
|
|
|
|
"github.com/oy/openyield/x/bearers/types"
|
|
)
|
|
|
|
// msg_server.go implements the bearers module's MsgServer (G-023 ownership
|
|
// split: cosmos-engineer scaffolds the file structure + method signatures;
|
|
// mesh-engineer/backend-engineer implements the handler logic bodies). The
|
|
// MsgServer wraps the Keeper + the BreadKeeper expected-keeper shim (already
|
|
// on the Keeper).
|
|
//
|
|
// Each method returns a (*Response, error). Handler state-machine ordering
|
|
// is enforced: ValidateBasic → keeper authz → state mutation →
|
|
// ctx.EventManager().EmitEvent.
|
|
//
|
|
// Surveillance-resistant invariant (A-522): NO handler emits geolocation or
|
|
// sender physical location. The surveillance-resistant locked const on
|
|
// OYSATLink/OYLRLink is a runtime invariant — a handler that emits
|
|
// geolocation violates it. A negative simtest asserts the event set
|
|
// contains NO geolocation fields.
|
|
//
|
|
// One-shot OY-QR (A-521): the MsgConsumeOYQR handler flips consumed BEFORE
|
|
// the transfer effect (state write FIRST, then the BreadKeeper shim call).
|
|
// A replay finds consumed==true and returns an error (idempotent reject,
|
|
// NOT double-effect). The SDK store is atomic per tx — a panic in the
|
|
// transfer rolls back the whole tx, so the order is safe; the order
|
|
// documents intent and matches the ibc-go delete-before-mint convention.
|
|
|
|
// msgServer is the concrete MsgServer implementation wrapping the Keeper.
|
|
type msgServer struct {
|
|
Keeper
|
|
}
|
|
|
|
// NewMsgServerImpl returns the bearers MsgServer for the provided Keeper.
|
|
func NewMsgServerImpl(k Keeper) types.MsgServer {
|
|
return &msgServer{Keeper: k}
|
|
}
|
|
|
|
var _ types.MsgServer = msgServer{}
|
|
|
|
// unwrapCtx extracts the sdk.Context from the interface-typed ctx.
|
|
func unwrapCtx(ctx interface{}) sdk.Context {
|
|
if c, ok := ctx.(sdk.Context); ok {
|
|
return c
|
|
}
|
|
panic(fmt.Sprintf("bearers: expected sdk.Context, got %T", ctx))
|
|
}
|
|
|
|
// nowUnix returns the current block time as unix seconds from the ctx.
|
|
func nowUnix(ctx sdk.Context) int64 {
|
|
return ctx.BlockTime().Unix()
|
|
}
|
|
|
|
// --- OpenSession (creates Session status=Open) -------------------------------
|
|
|
|
// OpenSession creates a new Session with status=Open. ValidateBasic is
|
|
// stateless; the handler enforces idempotency (session-id must not already
|
|
// exist).
|
|
func (s msgServer) OpenSession(ctx interface{}, msg *types.MsgOpenSession) (*types.MsgOpenSessionResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
// Idempotency: session-id must not already exist.
|
|
if _, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID); ok {
|
|
return nil, fmt.Errorf("bearers: session %q already exists", msg.SessionID)
|
|
}
|
|
|
|
session := types.Session{
|
|
SessionID: msg.SessionID,
|
|
BearerType: msg.BearerType,
|
|
InitiatorReach: msg.InitiatorReach,
|
|
PeerReach: msg.PeerReach,
|
|
Status: types.SessionOpen,
|
|
Frames: []types.Frame{},
|
|
TTL: msg.TTL,
|
|
OpenedAt: nowUnix(sdkCtx),
|
|
}
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.session_opened",
|
|
sdk.NewAttribute("session_id", msg.SessionID),
|
|
sdk.NewAttribute("bearer_type", string(msg.BearerType)),
|
|
sdk.NewAttribute("initiator_reach", msg.InitiatorReach),
|
|
sdk.NewAttribute("peer_reach", msg.PeerReach),
|
|
sdk.NewAttribute("status", string(types.SessionOpen)),
|
|
// NO geolocation (A-522 surveillance-resistant invariant).
|
|
))
|
|
return &types.MsgOpenSessionResponse{}, nil
|
|
}
|
|
|
|
// --- CloseSession (Active → Closed) ------------------------------------------
|
|
|
|
// CloseSession transitions an Active session to Closed. The handler
|
|
// enforces the stateful source-status check (must be Open or Active; an
|
|
// Open session with no frames can close directly).
|
|
func (s msgServer) CloseSession(ctx interface{}, msg *types.MsgCloseSession) (*types.MsgCloseSessionResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID)
|
|
}
|
|
if session.IsTerminal() {
|
|
return nil, fmt.Errorf("bearers: session %q is terminal (%s), cannot close", msg.SessionID, session.Status)
|
|
}
|
|
|
|
session.Status = types.SessionClosed
|
|
session.ClosedAt = nowUnix(sdkCtx)
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.session_closed",
|
|
sdk.NewAttribute("session_id", msg.SessionID),
|
|
sdk.NewAttribute("status", string(types.SessionClosed)),
|
|
))
|
|
return &types.MsgCloseSessionResponse{}, nil
|
|
}
|
|
|
|
// --- RevokeSession (out-of-band → Revoked) -----------------------------------
|
|
|
|
// RevokeSession transitions a session to Revoked (out-of-band termination).
|
|
// A revoked session rejects further Receive. The handler enforces the
|
|
// stateful source-status check (must not already be terminal).
|
|
func (s msgServer) RevokeSession(ctx interface{}, msg *types.MsgRevokeSession) (*types.MsgRevokeSessionResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID)
|
|
}
|
|
if session.IsTerminal() {
|
|
return nil, fmt.Errorf("bearers: session %q is terminal (%s), cannot revoke", msg.SessionID, session.Status)
|
|
}
|
|
|
|
session.Status = types.SessionRevoked
|
|
session.ClosedAt = nowUnix(sdkCtx)
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.session_revoked",
|
|
sdk.NewAttribute("session_id", msg.SessionID),
|
|
sdk.NewAttribute("status", string(types.SessionRevoked)),
|
|
))
|
|
return &types.MsgRevokeSessionResponse{}, nil
|
|
}
|
|
|
|
// --- SendOYSATFrame (send a frame on an Open/Active session) ------------------
|
|
|
|
// SendOYSATFrame sends a frame on an OY-SAT session. The handler enforces
|
|
// the stateful session-status check: the session must be Open or Active
|
|
// (frames on Closed/Revoked are REJECTED — the rejected-frame case).
|
|
func (s msgServer) SendOYSATFrame(ctx interface{}, msg *types.MsgSendOYSATFrame) (*types.MsgSendOYSATFrameResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID)
|
|
}
|
|
if session.IsTerminal() {
|
|
// Rejected-frame case: a frame received on a Closed/Revoked
|
|
// session MUST be rejected (A-523 session state machine).
|
|
return nil, fmt.Errorf("bearers: session %q is terminal (%s), rejects frame", msg.SessionID, session.Status)
|
|
}
|
|
if session.IsExpired(nowUnix(sdkCtx)) {
|
|
// TTL expiry transitions the session to Closed (the handler
|
|
// enforces expiry on Send/Receive checks).
|
|
session.Status = types.SessionClosed
|
|
session.ClosedAt = nowUnix(sdkCtx)
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
return nil, fmt.Errorf("bearers: session %q expired (ttl %d), rejects frame", msg.SessionID, session.TTL)
|
|
}
|
|
|
|
frame := types.Frame{
|
|
FrameID: msg.FrameID,
|
|
SenderReach: msg.Signer,
|
|
PayloadBytes: msg.PayloadBytes,
|
|
SentAt: nowUnix(sdkCtx),
|
|
}
|
|
session.Frames = append(session.Frames, frame)
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.frame_sent",
|
|
sdk.NewAttribute("session_id", msg.SessionID),
|
|
sdk.NewAttribute("frame_id", msg.FrameID),
|
|
sdk.NewAttribute("sender_reach", msg.Signer),
|
|
// NO geolocation (A-522 surveillance-resistant invariant).
|
|
))
|
|
return &types.MsgSendOYSATFrameResponse{}, nil
|
|
}
|
|
|
|
// --- ReceiveOYSATFrame (ack a frame; Open → Active on first ack) -------------
|
|
|
|
// ReceiveOYSATFrame acknowledges receipt of an OY-SAT frame. The handler
|
|
// transitions the session Open → Active on the first ack. The handler
|
|
// enforces the stateful session-status check: the session must be Open or
|
|
// Active (acks on Closed/Revoked are REJECTED — the rejected-frame case).
|
|
func (s msgServer) ReceiveOYSATFrame(ctx interface{}, msg *types.MsgReceiveOYSATFrame) (*types.MsgReceiveOYSATFrameResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID)
|
|
}
|
|
if session.IsTerminal() {
|
|
// Rejected-frame case: an ack received on a Closed/Revoked
|
|
// session MUST be rejected (A-523 session state machine).
|
|
return nil, fmt.Errorf("bearers: session %q is terminal (%s), rejects ack", msg.SessionID, session.Status)
|
|
}
|
|
if session.IsExpired(nowUnix(sdkCtx)) {
|
|
session.Status = types.SessionClosed
|
|
session.ClosedAt = nowUnix(sdkCtx)
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
return nil, fmt.Errorf("bearers: session %q expired (ttl %d), rejects ack", msg.SessionID, session.TTL)
|
|
}
|
|
|
|
// Find the named frame; mark it received.
|
|
found := false
|
|
for i := range session.Frames {
|
|
if session.Frames[i].FrameID == msg.FrameID {
|
|
session.Frames[i].Received = true
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found {
|
|
return nil, fmt.Errorf("bearers: frame %q not found on session %q", msg.FrameID, msg.SessionID)
|
|
}
|
|
|
|
// Open → Active on the first ack.
|
|
if session.Status == types.SessionOpen {
|
|
session.Status = types.SessionActive
|
|
}
|
|
s.Keeper.SetSession(sdkCtx, session)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.frame_received",
|
|
sdk.NewAttribute("session_id", msg.SessionID),
|
|
sdk.NewAttribute("frame_id", msg.FrameID),
|
|
sdk.NewAttribute("status", string(session.Status)),
|
|
// NO geolocation (A-522 surveillance-resistant invariant).
|
|
))
|
|
return &types.MsgReceiveOYSATFrameResponse{}, nil
|
|
}
|
|
|
|
// --- IssueOYQR (issue a one-shot OY-QR, consumed=false) ----------------------
|
|
|
|
// IssueOYQR issues a one-shot OY-QR (consumed=false). The handler enforces
|
|
// idempotency (qr-id must not already exist).
|
|
func (s msgServer) IssueOYQR(ctx interface{}, msg *types.MsgIssueOYQR) (*types.MsgIssueOYQRResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
// Idempotency: qr-id must not already exist.
|
|
if _, ok := s.Keeper.GetOYQRCode(sdkCtx, msg.QRID); ok {
|
|
return nil, fmt.Errorf("bearers: qr %q already exists", msg.QRID)
|
|
}
|
|
|
|
qr := types.OYQRCode{
|
|
QRID: msg.QRID,
|
|
PayloadBytes: msg.PayloadBytes,
|
|
Consumed: false,
|
|
IssuerReachID: msg.IssuerReachID,
|
|
AmountGrain: msg.AmountGrain,
|
|
ExpiresAt: msg.ExpiresAt,
|
|
}
|
|
s.Keeper.SetOYQRCode(sdkCtx, qr)
|
|
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.qr_issued",
|
|
sdk.NewAttribute("qr_id", msg.QRID),
|
|
sdk.NewAttribute("issuer_reach", msg.IssuerReachID),
|
|
sdk.NewAttribute("amount_grain", fmt.Sprintf("%d", msg.AmountGrain)),
|
|
sdk.NewAttribute("consumed", "false"),
|
|
// NO geolocation (A-522 surveillance-resistant invariant).
|
|
))
|
|
return &types.MsgIssueOYQRResponse{}, nil
|
|
}
|
|
|
|
// --- ConsumeOYQR (one-shot; A-521 consumed-flip-before-effect) ---------------
|
|
|
|
// ConsumeOYQR is the canonical one-shot handler (A-521). The ordering is:
|
|
// 1. load QR
|
|
// 2. assert !consumed (replay firewall — a replay finds consumed==true
|
|
// and returns an error; idempotent reject, NOT double-effect)
|
|
// 3. assert expires-at > now (the QR is still valid)
|
|
// 4. FLIP consumed=true (state write FIRST — A-521)
|
|
// 5. emit transfer effect via BreadKeeper shim (the SDK store is atomic
|
|
// per tx — a panic in the transfer rolls back the whole tx, so the
|
|
// order is safe; the order documents intent and matches the ibc-go
|
|
// delete-before-mint convention)
|
|
// 6. emit event
|
|
// 7. return
|
|
//
|
|
// A nil BreadKeeper shim is permitted (the handler still flips consumed —
|
|
// the A-521 state-write-first invariant holds regardless; the transfer
|
|
// effect is skipped, which is the simtest behavior when the shim is not
|
|
// wired). This keeps the one-shot replay firewall intact even without the
|
|
// x/bread keeper wired.
|
|
func (s msgServer) ConsumeOYQR(ctx interface{}, msg *types.MsgConsumeOYQR) (*types.MsgConsumeOYQRResponse, error) {
|
|
if err := msg.ValidateBasic(); err != nil {
|
|
return nil, err
|
|
}
|
|
sdkCtx := unwrapCtx(ctx)
|
|
|
|
// 1. Load QR.
|
|
qr, ok := s.Keeper.GetOYQRCode(sdkCtx, msg.QRID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("bearers: qr %q not found", msg.QRID)
|
|
}
|
|
|
|
// 2. Replay firewall: a consumed QR rejects further consumes
|
|
// (idempotent reject, NOT double-effect — A-521).
|
|
if qr.Consumed {
|
|
return nil, fmt.Errorf("bearers: qr %q already consumed (one-shot — A-521)", msg.QRID)
|
|
}
|
|
|
|
// 3. Expiry check: the QR must still be valid (expires-at > now).
|
|
now := nowUnix(sdkCtx)
|
|
if qr.ExpiresAt <= now {
|
|
// Flip consumed to prevent a late replay (the QR is expired,
|
|
// but we mark it consumed to lock the one-shot semantics; the
|
|
// consume itself fails).
|
|
qr.Consumed = true
|
|
s.Keeper.SetOYQRCode(sdkCtx, qr)
|
|
return nil, fmt.Errorf("bearers: qr %q expired (expires-at %d <= now %d)", msg.QRID, qr.ExpiresAt, now)
|
|
}
|
|
|
|
// 4. FLIP consumed=true (state write FIRST — A-521). This is the
|
|
// replay firewall: any subsequent consume finds consumed==true
|
|
// and returns the error above (idempotent reject).
|
|
qr.Consumed = true
|
|
s.Keeper.SetOYQRCode(sdkCtx, qr)
|
|
|
|
// 5. Emit transfer effect via BreadKeeper shim. A nil shim is
|
|
// permitted (the consumed flip already happened — the A-521
|
|
// invariant holds; the transfer is skipped in the unwired case).
|
|
var transferErr error
|
|
if s.Keeper.breadKeeper != nil {
|
|
transferErr = s.Keeper.breadKeeper.TransferGrain(qr.IssuerReachID, msg.ConsumerReachID, qr.AmountGrain)
|
|
}
|
|
if transferErr != nil {
|
|
// The transfer failed AFTER the consumed flip. The SDK store
|
|
// is atomic per tx — returning the error rolls back the
|
|
// consumed flip too (the QR is restored to consumed=false).
|
|
// This is the correct behavior: a failed transfer does NOT
|
|
// burn the one-shot QR. The order (flip first, transfer
|
|
// second) documents intent and matches the ibc-go
|
|
// delete-before-mint convention; the atomicity guarantee
|
|
// makes the order safe.
|
|
return nil, fmt.Errorf("bearers: qr %q transfer effect failed: %w", msg.QRID, transferErr)
|
|
}
|
|
|
|
// 6. Emit event.
|
|
sdkCtx.EventManager().EmitEvent(sdk.NewEvent(
|
|
"bearers.qr_consumed",
|
|
sdk.NewAttribute("qr_id", msg.QRID),
|
|
sdk.NewAttribute("issuer_reach", qr.IssuerReachID),
|
|
sdk.NewAttribute("consumer_reach", msg.ConsumerReachID),
|
|
sdk.NewAttribute("amount_grain", fmt.Sprintf("%d", qr.AmountGrain)),
|
|
sdk.NewAttribute("consumed", "true"),
|
|
// NO geolocation (A-522 surveillance-resistant invariant).
|
|
))
|
|
return &types.MsgConsumeOYQRResponse{}, nil
|
|
}
|