diff --git a/cln_plugin/cln_plugin.go b/cln_plugin/cln_plugin.go new file mode 100644 index 0000000..3a8442f --- /dev/null +++ b/cln_plugin/cln_plugin.go @@ -0,0 +1,117 @@ +package cln_plugin + +import ( + "encoding/hex" + "log" + "os" + + "github.com/breez/lspd/basetypes" + "github.com/niftynei/glightning/glightning" +) + +type ClnPlugin struct { + server *server + plugin *glightning.Plugin +} + +func NewClnPlugin(server *server) *ClnPlugin { + c := &ClnPlugin{ + server: server, + } + + c.plugin = glightning.NewPlugin(c.onInit) + c.plugin.RegisterHooks(&glightning.Hooks{ + HtlcAccepted: c.onHtlcAccepted, + }) + + return c +} + +func (c *ClnPlugin) Start() error { + err := c.plugin.Start(os.Stdin, os.Stdout) + if err != nil { + log.Printf("Plugin error: %v", err) + return err + } + + return nil +} + +func (c *ClnPlugin) Stop() { + c.plugin.Stop() + c.server.Stop() +} + +func (c *ClnPlugin) onInit(plugin *glightning.Plugin, options map[string]glightning.Option, config *glightning.Config) { + log.Printf("successfully init'd! %v\n", config.RpcFile) + + log.Printf("Starting htlc grpc server.") + go func() { + err := c.server.Start() + if err == nil { + log.Printf("WARNING server stopped.") + } else { + log.Printf("ERROR Server stopped with error: %v", err) + } + }() + + //lightning server + clientcln := glightning.NewLightning() + clientcln.SetTimeout(60) + clientcln.StartUp(config.RpcFile, config.LightningDir) + + log.Printf("successfull clientcln.StartUp") +} + +func (c *ClnPlugin) onHtlcAccepted(event *glightning.HtlcAcceptedEvent) (*glightning.HtlcAcceptedResponse, error) { + payload, err := hex.DecodeString(event.Onion.Payload) + if err != nil { + log.Printf("ERROR failed to decode payload %s: %v", event.Onion.Payload, err) + return nil, err + } + scid, err := basetypes.NewShortChannelIDFromString(event.Onion.ShortChannelId) + if err != nil { + log.Printf("ERROR failed to decode short channel id %s: %v", event.Onion.ShortChannelId, err) + return nil, err + } + ph, err := hex.DecodeString(event.Htlc.PaymentHash) + if err != nil { + log.Printf("ERROR failed to decode payment hash %s: %v", event.Onion.ShortChannelId, err) + return nil, err + } + + resp := c.server.Send(&HtlcAccepted{ + Onion: &Onion{ + Payload: payload, + ShortChannelId: uint64(*scid), + ForwardAmountMsat: event.Onion.ForwardAmount, + }, + Htlc: &HtlcOffer{ + AmountMsat: event.Htlc.AmountMilliSatoshi, + CltvExpiryRelative: uint32(event.Htlc.CltvExpiryRelative), + CltvExpiry: uint32(event.Htlc.CltvExpiry), + PaymentHash: ph, + }, + }) + + _, ok := resp.Outcome.(*HtlcResolution_Continue) + if ok { + return event.Continue(), nil + } + + cont, ok := resp.Outcome.(*HtlcResolution_ContinueWith) + if ok { + chanId := hex.EncodeToString(cont.ContinueWith.ChannelId) + pl := hex.EncodeToString(cont.ContinueWith.Payload) + return event.ContinueWith(chanId, pl), nil + } + + fail, ok := resp.Outcome.(*HtlcResolution_Fail) + if ok { + fm := hex.EncodeToString(fail.Fail.FailureMessage) + return event.Fail(fm), err + } + + log.Printf("Unexpected htlc resolution type %T: %+v", resp.Outcome, resp.Outcome) + return event.Fail("1007"), nil // temporary channel failure +} diff --git a/cln_plugin/cln_plugin.pb.go b/cln_plugin/cln_plugin.pb.go new file mode 100644 index 0000000..181a0f0 --- /dev/null +++ b/cln_plugin/cln_plugin.pb.go @@ -0,0 +1,674 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.28.1 +// protoc v3.21.8 +// source: cln_plugin.proto + +package cln_plugin + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type HtlcAccepted struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Correlationid uint64 `protobuf:"varint,1,opt,name=correlationid,proto3" json:"correlationid,omitempty"` + Onion *Onion `protobuf:"bytes,2,opt,name=onion,proto3" json:"onion,omitempty"` + Htlc *HtlcOffer `protobuf:"bytes,3,opt,name=htlc,proto3" json:"htlc,omitempty"` +} + +func (x *HtlcAccepted) Reset() { + *x = HtlcAccepted{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcAccepted) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcAccepted) ProtoMessage() {} + +func (x *HtlcAccepted) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcAccepted.ProtoReflect.Descriptor instead. +func (*HtlcAccepted) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{0} +} + +func (x *HtlcAccepted) GetCorrelationid() uint64 { + if x != nil { + return x.Correlationid + } + return 0 +} + +func (x *HtlcAccepted) GetOnion() *Onion { + if x != nil { + return x.Onion + } + return nil +} + +func (x *HtlcAccepted) GetHtlc() *HtlcOffer { + if x != nil { + return x.Htlc + } + return nil +} + +type Onion struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Payload []byte `protobuf:"bytes,1,opt,name=payload,proto3" json:"payload,omitempty"` + ShortChannelId uint64 `protobuf:"varint,2,opt,name=short_channel_id,json=shortChannelId,proto3" json:"short_channel_id,omitempty"` + ForwardAmountMsat uint64 `protobuf:"varint,3,opt,name=forward_amount_msat,json=forwardAmountMsat,proto3" json:"forward_amount_msat,omitempty"` +} + +func (x *Onion) Reset() { + *x = Onion{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Onion) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Onion) ProtoMessage() {} + +func (x *Onion) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Onion.ProtoReflect.Descriptor instead. +func (*Onion) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{1} +} + +func (x *Onion) GetPayload() []byte { + if x != nil { + return x.Payload + } + return nil +} + +func (x *Onion) GetShortChannelId() uint64 { + if x != nil { + return x.ShortChannelId + } + return 0 +} + +func (x *Onion) GetForwardAmountMsat() uint64 { + if x != nil { + return x.ForwardAmountMsat + } + return 0 +} + +type HtlcOffer struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + AmountMsat uint64 `protobuf:"varint,2,opt,name=amount_msat,json=amountMsat,proto3" json:"amount_msat,omitempty"` + CltvExpiryRelative uint32 `protobuf:"varint,3,opt,name=cltv_expiry_relative,json=cltvExpiryRelative,proto3" json:"cltv_expiry_relative,omitempty"` + CltvExpiry uint32 `protobuf:"varint,4,opt,name=cltv_expiry,json=cltvExpiry,proto3" json:"cltv_expiry,omitempty"` + PaymentHash []byte `protobuf:"bytes,5,opt,name=payment_hash,json=paymentHash,proto3" json:"payment_hash,omitempty"` +} + +func (x *HtlcOffer) Reset() { + *x = HtlcOffer{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcOffer) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcOffer) ProtoMessage() {} + +func (x *HtlcOffer) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcOffer.ProtoReflect.Descriptor instead. +func (*HtlcOffer) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{2} +} + +func (x *HtlcOffer) GetAmountMsat() uint64 { + if x != nil { + return x.AmountMsat + } + return 0 +} + +func (x *HtlcOffer) GetCltvExpiryRelative() uint32 { + if x != nil { + return x.CltvExpiryRelative + } + return 0 +} + +func (x *HtlcOffer) GetCltvExpiry() uint32 { + if x != nil { + return x.CltvExpiry + } + return 0 +} + +func (x *HtlcOffer) GetPaymentHash() []byte { + if x != nil { + return x.PaymentHash + } + return nil +} + +type HtlcResolution struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Correlationid uint64 `protobuf:"varint,1,opt,name=correlationid,proto3" json:"correlationid,omitempty"` + // Types that are assignable to Outcome: + // *HtlcResolution_Fail + // *HtlcResolution_Continue + // *HtlcResolution_ContinueWith + Outcome isHtlcResolution_Outcome `protobuf_oneof:"outcome"` +} + +func (x *HtlcResolution) Reset() { + *x = HtlcResolution{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcResolution) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcResolution) ProtoMessage() {} + +func (x *HtlcResolution) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcResolution.ProtoReflect.Descriptor instead. +func (*HtlcResolution) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{3} +} + +func (x *HtlcResolution) GetCorrelationid() uint64 { + if x != nil { + return x.Correlationid + } + return 0 +} + +func (m *HtlcResolution) GetOutcome() isHtlcResolution_Outcome { + if m != nil { + return m.Outcome + } + return nil +} + +func (x *HtlcResolution) GetFail() *HtlcFail { + if x, ok := x.GetOutcome().(*HtlcResolution_Fail); ok { + return x.Fail + } + return nil +} + +func (x *HtlcResolution) GetContinue() *HtlcContinue { + if x, ok := x.GetOutcome().(*HtlcResolution_Continue); ok { + return x.Continue + } + return nil +} + +func (x *HtlcResolution) GetContinueWith() *HtlcContinueWith { + if x, ok := x.GetOutcome().(*HtlcResolution_ContinueWith); ok { + return x.ContinueWith + } + return nil +} + +type isHtlcResolution_Outcome interface { + isHtlcResolution_Outcome() +} + +type HtlcResolution_Fail struct { + Fail *HtlcFail `protobuf:"bytes,2,opt,name=fail,proto3,oneof"` +} + +type HtlcResolution_Continue struct { + Continue *HtlcContinue `protobuf:"bytes,3,opt,name=continue,proto3,oneof"` +} + +type HtlcResolution_ContinueWith struct { + ContinueWith *HtlcContinueWith `protobuf:"bytes,4,opt,name=continue_with,json=continueWith,proto3,oneof"` +} + +func (*HtlcResolution_Fail) isHtlcResolution_Outcome() {} + +func (*HtlcResolution_Continue) isHtlcResolution_Outcome() {} + +func (*HtlcResolution_ContinueWith) isHtlcResolution_Outcome() {} + +type HtlcFail struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + FailureMessage []byte `protobuf:"bytes,1,opt,name=failure_message,json=failureMessage,proto3" json:"failure_message,omitempty"` +} + +func (x *HtlcFail) Reset() { + *x = HtlcFail{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcFail) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcFail) ProtoMessage() {} + +func (x *HtlcFail) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcFail.ProtoReflect.Descriptor instead. +func (*HtlcFail) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{4} +} + +func (x *HtlcFail) GetFailureMessage() []byte { + if x != nil { + return x.FailureMessage + } + return nil +} + +type HtlcContinue struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *HtlcContinue) Reset() { + *x = HtlcContinue{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcContinue) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcContinue) ProtoMessage() {} + +func (x *HtlcContinue) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[5] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcContinue.ProtoReflect.Descriptor instead. +func (*HtlcContinue) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{5} +} + +type HtlcContinueWith struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + ChannelId []byte `protobuf:"bytes,1,opt,name=channel_id,json=channelId,proto3" json:"channel_id,omitempty"` + Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` +} + +func (x *HtlcContinueWith) Reset() { + *x = HtlcContinueWith{} + if protoimpl.UnsafeEnabled { + mi := &file_cln_plugin_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HtlcContinueWith) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HtlcContinueWith) ProtoMessage() {} + +func (x *HtlcContinueWith) ProtoReflect() protoreflect.Message { + mi := &file_cln_plugin_proto_msgTypes[6] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HtlcContinueWith.ProtoReflect.Descriptor instead. +func (*HtlcContinueWith) Descriptor() ([]byte, []int) { + return file_cln_plugin_proto_rawDescGZIP(), []int{6} +} + +func (x *HtlcContinueWith) GetChannelId() []byte { + if x != nil { + return x.ChannelId + } + return nil +} + +func (x *HtlcContinueWith) GetPayload() []byte { + if x != nil { + return x.Payload + } + return nil +} + +var File_cln_plugin_proto protoreflect.FileDescriptor + +var file_cln_plugin_proto_rawDesc = []byte{ + 0x0a, 0x10, 0x63, 0x6c, 0x6e, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x22, 0x72, 0x0a, 0x0c, 0x48, 0x74, 0x6c, 0x63, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, + 0x65, 0x64, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f, 0x72, 0x72, 0x65, 0x6c, 0x61, 0x74, 0x69, 0x6f, + 0x6e, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x63, 0x6f, 0x72, 0x72, 0x65, + 0x6c, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x05, 0x6f, 0x6e, 0x69, 0x6f, + 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x4f, 0x6e, 0x69, 0x6f, 0x6e, 0x52, + 0x05, 0x6f, 0x6e, 0x69, 0x6f, 0x6e, 0x12, 0x1e, 0x0a, 0x04, 0x68, 0x74, 0x6c, 0x63, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x4f, 0x66, 0x66, 0x65, 0x72, + 0x52, 0x04, 0x68, 0x74, 0x6c, 0x63, 0x22, 0x7b, 0x0a, 0x05, 0x4f, 0x6e, 0x69, 0x6f, 0x6e, 0x12, + 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, + 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x73, 0x68, 0x6f, + 0x72, 0x74, 0x5f, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, + 0x01, 0x28, 0x04, 0x52, 0x0e, 0x73, 0x68, 0x6f, 0x72, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, + 0x6c, 0x49, 0x64, 0x12, 0x2e, 0x0a, 0x13, 0x66, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x5f, 0x61, + 0x6d, 0x6f, 0x75, 0x6e, 0x74, 0x5f, 0x6d, 0x73, 0x61, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, + 0x52, 0x11, 0x66, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x41, 0x6d, 0x6f, 0x75, 0x6e, 0x74, 0x4d, + 0x73, 0x61, 0x74, 0x22, 0xa2, 0x01, 0x0a, 0x09, 0x48, 0x74, 0x6c, 0x63, 0x4f, 0x66, 0x66, 0x65, + 0x72, 0x12, 0x1f, 0x0a, 0x0b, 0x61, 0x6d, 0x6f, 0x75, 0x6e, 0x74, 0x5f, 0x6d, 0x73, 0x61, 0x74, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x61, 0x6d, 0x6f, 0x75, 0x6e, 0x74, 0x4d, 0x73, + 0x61, 0x74, 0x12, 0x30, 0x0a, 0x14, 0x63, 0x6c, 0x74, 0x76, 0x5f, 0x65, 0x78, 0x70, 0x69, 0x72, + 0x79, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x74, 0x69, 0x76, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, + 0x52, 0x12, 0x63, 0x6c, 0x74, 0x76, 0x45, 0x78, 0x70, 0x69, 0x72, 0x79, 0x52, 0x65, 0x6c, 0x61, + 0x74, 0x69, 0x76, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x63, 0x6c, 0x74, 0x76, 0x5f, 0x65, 0x78, 0x70, + 0x69, 0x72, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x63, 0x6c, 0x74, 0x76, 0x45, + 0x78, 0x70, 0x69, 0x72, 0x79, 0x12, 0x21, 0x0a, 0x0c, 0x70, 0x61, 0x79, 0x6d, 0x65, 0x6e, 0x74, + 0x5f, 0x68, 0x61, 0x73, 0x68, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x70, 0x61, 0x79, + 0x6d, 0x65, 0x6e, 0x74, 0x48, 0x61, 0x73, 0x68, 0x22, 0xc9, 0x01, 0x0a, 0x0e, 0x48, 0x74, 0x6c, + 0x63, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x24, 0x0a, 0x0d, 0x63, + 0x6f, 0x72, 0x72, 0x65, 0x6c, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x04, 0x52, 0x0d, 0x63, 0x6f, 0x72, 0x72, 0x65, 0x6c, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x69, + 0x64, 0x12, 0x1f, 0x0a, 0x04, 0x66, 0x61, 0x69, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, + 0x09, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x46, 0x61, 0x69, 0x6c, 0x48, 0x00, 0x52, 0x04, 0x66, 0x61, + 0x69, 0x6c, 0x12, 0x2b, 0x0a, 0x08, 0x63, 0x6f, 0x6e, 0x74, 0x69, 0x6e, 0x75, 0x65, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x43, 0x6f, 0x6e, 0x74, 0x69, + 0x6e, 0x75, 0x65, 0x48, 0x00, 0x52, 0x08, 0x63, 0x6f, 0x6e, 0x74, 0x69, 0x6e, 0x75, 0x65, 0x12, + 0x38, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x69, 0x6e, 0x75, 0x65, 0x5f, 0x77, 0x69, 0x74, 0x68, + 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x43, 0x6f, 0x6e, + 0x74, 0x69, 0x6e, 0x75, 0x65, 0x57, 0x69, 0x74, 0x68, 0x48, 0x00, 0x52, 0x0c, 0x63, 0x6f, 0x6e, + 0x74, 0x69, 0x6e, 0x75, 0x65, 0x57, 0x69, 0x74, 0x68, 0x42, 0x09, 0x0a, 0x07, 0x6f, 0x75, 0x74, + 0x63, 0x6f, 0x6d, 0x65, 0x22, 0x33, 0x0a, 0x08, 0x48, 0x74, 0x6c, 0x63, 0x46, 0x61, 0x69, 0x6c, + 0x12, 0x27, 0x0a, 0x0f, 0x66, 0x61, 0x69, 0x6c, 0x75, 0x72, 0x65, 0x5f, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0e, 0x66, 0x61, 0x69, 0x6c, 0x75, + 0x72, 0x65, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x0e, 0x0a, 0x0c, 0x48, 0x74, 0x6c, + 0x63, 0x43, 0x6f, 0x6e, 0x74, 0x69, 0x6e, 0x75, 0x65, 0x22, 0x4b, 0x0a, 0x10, 0x48, 0x74, 0x6c, + 0x63, 0x43, 0x6f, 0x6e, 0x74, 0x69, 0x6e, 0x75, 0x65, 0x57, 0x69, 0x74, 0x68, 0x12, 0x1d, 0x0a, + 0x0a, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x0c, 0x52, 0x09, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, + 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, + 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x32, 0x3d, 0x0a, 0x09, 0x43, 0x6c, 0x6e, 0x50, 0x6c, 0x75, + 0x67, 0x69, 0x6e, 0x12, 0x30, 0x0a, 0x0a, 0x48, 0x74, 0x6c, 0x63, 0x53, 0x74, 0x72, 0x65, 0x61, + 0x6d, 0x12, 0x0f, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x75, 0x74, 0x69, + 0x6f, 0x6e, 0x1a, 0x0d, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, 0x65, + 0x64, 0x28, 0x01, 0x30, 0x01, 0x42, 0x22, 0x5a, 0x20, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, + 0x63, 0x6f, 0x6d, 0x2f, 0x62, 0x72, 0x65, 0x65, 0x7a, 0x2f, 0x6c, 0x73, 0x70, 0x64, 0x2f, 0x63, + 0x6c, 0x6e, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x33, +} + +var ( + file_cln_plugin_proto_rawDescOnce sync.Once + file_cln_plugin_proto_rawDescData = file_cln_plugin_proto_rawDesc +) + +func file_cln_plugin_proto_rawDescGZIP() []byte { + file_cln_plugin_proto_rawDescOnce.Do(func() { + file_cln_plugin_proto_rawDescData = protoimpl.X.CompressGZIP(file_cln_plugin_proto_rawDescData) + }) + return file_cln_plugin_proto_rawDescData +} + +var file_cln_plugin_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_cln_plugin_proto_goTypes = []interface{}{ + (*HtlcAccepted)(nil), // 0: HtlcAccepted + (*Onion)(nil), // 1: Onion + (*HtlcOffer)(nil), // 2: HtlcOffer + (*HtlcResolution)(nil), // 3: HtlcResolution + (*HtlcFail)(nil), // 4: HtlcFail + (*HtlcContinue)(nil), // 5: HtlcContinue + (*HtlcContinueWith)(nil), // 6: HtlcContinueWith +} +var file_cln_plugin_proto_depIdxs = []int32{ + 1, // 0: HtlcAccepted.onion:type_name -> Onion + 2, // 1: HtlcAccepted.htlc:type_name -> HtlcOffer + 4, // 2: HtlcResolution.fail:type_name -> HtlcFail + 5, // 3: HtlcResolution.continue:type_name -> HtlcContinue + 6, // 4: HtlcResolution.continue_with:type_name -> HtlcContinueWith + 3, // 5: ClnPlugin.HtlcStream:input_type -> HtlcResolution + 0, // 6: ClnPlugin.HtlcStream:output_type -> HtlcAccepted + 6, // [6:7] is the sub-list for method output_type + 5, // [5:6] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name +} + +func init() { file_cln_plugin_proto_init() } +func file_cln_plugin_proto_init() { + if File_cln_plugin_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_cln_plugin_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcAccepted); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Onion); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcOffer); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcResolution); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcFail); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcContinue); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_cln_plugin_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HtlcContinueWith); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + file_cln_plugin_proto_msgTypes[3].OneofWrappers = []interface{}{ + (*HtlcResolution_Fail)(nil), + (*HtlcResolution_Continue)(nil), + (*HtlcResolution_ContinueWith)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_cln_plugin_proto_rawDesc, + NumEnums: 0, + NumMessages: 7, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_cln_plugin_proto_goTypes, + DependencyIndexes: file_cln_plugin_proto_depIdxs, + MessageInfos: file_cln_plugin_proto_msgTypes, + }.Build() + File_cln_plugin_proto = out.File + file_cln_plugin_proto_rawDesc = nil + file_cln_plugin_proto_goTypes = nil + file_cln_plugin_proto_depIdxs = nil +} diff --git a/cln_plugin/cln_plugin_grpc.pb.go b/cln_plugin/cln_plugin_grpc.pb.go new file mode 100644 index 0000000..3d4c9f9 --- /dev/null +++ b/cln_plugin/cln_plugin_grpc.pb.go @@ -0,0 +1,137 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.2.0 +// - protoc v3.21.8 +// source: cln_plugin.proto + +package cln_plugin + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.32.0 or later. +const _ = grpc.SupportPackageIsVersion7 + +// ClnPluginClient is the client API for ClnPlugin service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type ClnPluginClient interface { + HtlcStream(ctx context.Context, opts ...grpc.CallOption) (ClnPlugin_HtlcStreamClient, error) +} + +type clnPluginClient struct { + cc grpc.ClientConnInterface +} + +func NewClnPluginClient(cc grpc.ClientConnInterface) ClnPluginClient { + return &clnPluginClient{cc} +} + +func (c *clnPluginClient) HtlcStream(ctx context.Context, opts ...grpc.CallOption) (ClnPlugin_HtlcStreamClient, error) { + stream, err := c.cc.NewStream(ctx, &ClnPlugin_ServiceDesc.Streams[0], "/ClnPlugin/HtlcStream", opts...) + if err != nil { + return nil, err + } + x := &clnPluginHtlcStreamClient{stream} + return x, nil +} + +type ClnPlugin_HtlcStreamClient interface { + Send(*HtlcResolution) error + Recv() (*HtlcAccepted, error) + grpc.ClientStream +} + +type clnPluginHtlcStreamClient struct { + grpc.ClientStream +} + +func (x *clnPluginHtlcStreamClient) Send(m *HtlcResolution) error { + return x.ClientStream.SendMsg(m) +} + +func (x *clnPluginHtlcStreamClient) Recv() (*HtlcAccepted, error) { + m := new(HtlcAccepted) + if err := x.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + +// ClnPluginServer is the server API for ClnPlugin service. +// All implementations must embed UnimplementedClnPluginServer +// for forward compatibility +type ClnPluginServer interface { + HtlcStream(ClnPlugin_HtlcStreamServer) error + mustEmbedUnimplementedClnPluginServer() +} + +// UnimplementedClnPluginServer must be embedded to have forward compatible implementations. +type UnimplementedClnPluginServer struct { +} + +func (UnimplementedClnPluginServer) HtlcStream(ClnPlugin_HtlcStreamServer) error { + return status.Errorf(codes.Unimplemented, "method HtlcStream not implemented") +} +func (UnimplementedClnPluginServer) mustEmbedUnimplementedClnPluginServer() {} + +// UnsafeClnPluginServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to ClnPluginServer will +// result in compilation errors. +type UnsafeClnPluginServer interface { + mustEmbedUnimplementedClnPluginServer() +} + +func RegisterClnPluginServer(s grpc.ServiceRegistrar, srv ClnPluginServer) { + s.RegisterService(&ClnPlugin_ServiceDesc, srv) +} + +func _ClnPlugin_HtlcStream_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(ClnPluginServer).HtlcStream(&clnPluginHtlcStreamServer{stream}) +} + +type ClnPlugin_HtlcStreamServer interface { + Send(*HtlcAccepted) error + Recv() (*HtlcResolution, error) + grpc.ServerStream +} + +type clnPluginHtlcStreamServer struct { + grpc.ServerStream +} + +func (x *clnPluginHtlcStreamServer) Send(m *HtlcAccepted) error { + return x.ServerStream.SendMsg(m) +} + +func (x *clnPluginHtlcStreamServer) Recv() (*HtlcResolution, error) { + m := new(HtlcResolution) + if err := x.ServerStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + +// ClnPlugin_ServiceDesc is the grpc.ServiceDesc for ClnPlugin service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var ClnPlugin_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "ClnPlugin", + HandlerType: (*ClnPluginServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "HtlcStream", + Handler: _ClnPlugin_HtlcStream_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "cln_plugin.proto", +} diff --git a/cln_plugin/cmd/main.go b/cln_plugin/cmd/main.go new file mode 100644 index 0000000..5a84e46 --- /dev/null +++ b/cln_plugin/cmd/main.go @@ -0,0 +1,31 @@ +package main + +import ( + "log" + "os" + "os/signal" + "syscall" + + "github.com/breez/lspd/cln_plugin" +) + +func main() { + listen := os.Getenv("LISTEN_ADDRESS") + server := cln_plugin.NewServer(listen) + plugin := cln_plugin.NewClnPlugin(server) + + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt, syscall.SIGINT) + go func() { + sig := <-c + log.Printf("Received stop signal %v. Stopping.", sig) + plugin.Stop() + }() + + err := plugin.Start() + if err == nil { + log.Printf("cln plugin stopped.") + } else { + log.Printf("cln plugin stopped with error: %v", err) + } +} diff --git a/cln_plugin/genproto.sh b/cln_plugin/genproto.sh new file mode 100755 index 0000000..226b0fe --- /dev/null +++ b/cln_plugin/genproto.sh @@ -0,0 +1,5 @@ +#!/bin/bash +SCRIPTDIR=$(dirname $0) +PROTO_ROOT=$SCRIPTDIR/proto + +protoc --go_out=$SCRIPTDIR --go_opt=paths=source_relative --go-grpc_out=$SCRIPTDIR --go-grpc_opt=paths=source_relative -I=$PROTO_ROOT $PROTO_ROOT/* \ No newline at end of file diff --git a/cln_plugin/proto/cln_plugin.proto b/cln_plugin/proto/cln_plugin.proto new file mode 100644 index 0000000..f284d40 --- /dev/null +++ b/cln_plugin/proto/cln_plugin.proto @@ -0,0 +1,46 @@ +syntax = "proto3"; +option go_package="github.com/breez/lspd/cln_plugin"; + +service ClnPlugin { + rpc HtlcStream(stream HtlcResolution) returns (stream HtlcAccepted); +} + +message HtlcAccepted { + uint64 correlationid = 1; + Onion onion = 2; + HtlcOffer htlc = 3; +} + +message Onion { + bytes payload = 1; + uint64 short_channel_id = 2; + uint64 forward_amount_msat = 3; +} + +message HtlcOffer { + uint64 amount_msat = 2; + uint32 cltv_expiry_relative = 3; + uint32 cltv_expiry = 4; + bytes payment_hash = 5; +} + +message HtlcResolution { + uint64 correlationid = 1; + oneof outcome { + HtlcFail fail = 2; + HtlcContinue continue = 3; + HtlcContinueWith continue_with = 4; + } +} + +message HtlcFail { + bytes failure_message = 1; +} + +message HtlcContinue { +} + +message HtlcContinueWith { + bytes channel_id = 1; + bytes payload = 2; +} \ No newline at end of file diff --git a/cln_plugin/sample.env b/cln_plugin/sample.env new file mode 100644 index 0000000..1dc0012 --- /dev/null +++ b/cln_plugin/sample.env @@ -0,0 +1 @@ +LISTEN_ADDRESS=
\ No newline at end of file diff --git a/cln_plugin/server.go b/cln_plugin/server.go new file mode 100644 index 0000000..b677b96 --- /dev/null +++ b/cln_plugin/server.go @@ -0,0 +1,225 @@ +package cln_plugin + +import ( + "fmt" + "io" + "log" + "net" + "sync" + "time" + + grpc "google.golang.org/grpc" +) + +var receiveWaitDelay = time.Millisecond * 200 + +type subscription struct { + stream ClnPlugin_HtlcStreamServer + done chan struct{} +} +type server struct { + ClnPluginServer + listenAddress string + grpcServer *grpc.Server + startMtx sync.Mutex + corrMtx sync.Mutex + subscription *subscription + newSubscriber chan struct{} + done chan struct{} + err chan error + correlations map[uint64]chan *HtlcResolution + index uint64 +} + +func NewServer(listenAddress string) *server { + return &server{ + listenAddress: listenAddress, + newSubscriber: make(chan struct{}, 1), + done: make(chan struct{}), + err: make(chan error, 1), + correlations: make(map[uint64]chan *HtlcResolution), + index: 0, + } +} + +func (s *server) Start() error { + s.startMtx.Lock() + if s.grpcServer != nil { + s.startMtx.Unlock() + return nil + } + + lis, err := net.Listen("tcp", s.listenAddress) + if err != nil { + log.Printf("ERROR Server failed to listen: %v", err) + s.startMtx.Unlock() + return err + } + + s.grpcServer = grpc.NewServer() + s.startMtx.Unlock() + RegisterClnPluginServer(s.grpcServer, s) + + log.Printf("Server starting to listen on %s.", s.listenAddress) + go s.listenHtlcResponses() + return s.grpcServer.Serve(lis) +} + +func (s *server) Stop() { + s.startMtx.Lock() + defer s.startMtx.Unlock() + log.Printf("Server Stop() called.") + if s.grpcServer == nil { + return + } + + close(s.done) + s.grpcServer.Stop() + s.grpcServer = nil +} + +func (s *server) HtlcStream(stream ClnPlugin_HtlcStreamServer) error { + log.Printf("Got HTLC stream subscription request.") + s.startMtx.Lock() + if s.subscription != nil { + s.startMtx.Unlock() + return fmt.Errorf("already subscribed") + } + + sb := &subscription{ + stream: stream, + done: make(chan struct{}), + } + s.subscription = sb + s.newSubscriber <- struct{}{} + s.startMtx.Unlock() + + defer func() { + s.startMtx.Lock() + s.subscription = nil + close(sb.done) + s.startMtx.Unlock() + }() + + go func() { + <-stream.Context().Done() + log.Printf("HtlcStream context is done. Removing subscriber: %v", stream.Context().Err()) + s.startMtx.Lock() + s.subscription = nil + close(sb.done) + s.startMtx.Unlock() + }() + + for { + select { + case <-s.done: + log.Printf("HTLC server signalled done. Return EOF.") + return io.EOF + case err := <-s.err: + log.Printf("HTLC server signalled error: %v", err) + return err + case <-sb.done: + log.Printf("HTLC stream signalled done. Return EOF.") + return io.EOF + } + } +} + +func (s *server) Send(h *HtlcAccepted) *HtlcResolution { + sb := s.subscription + if sb == nil { + log.Printf("No subscribers available. Ignoring HtlcAccepted %+v", h) + return s.defaultResolution() + } + + c := make(chan *HtlcResolution) + s.corrMtx.Lock() + s.index++ + index := s.index + s.correlations[index] = c + s.corrMtx.Unlock() + + h.Correlationid = index + + defer func() { + s.corrMtx.Lock() + delete(s.correlations, index) + s.corrMtx.Unlock() + close(c) + }() + + log.Printf("Sending HtlcAccepted: %+v", h) + err := sb.stream.Send(h) + if err != nil { + // TODO: Close the connection? Reset the subscriber? + log.Printf("Send() errored, Correlationid: %d: %v", index, err) + return s.defaultResolution() + } + + select { + case <-s.done: + log.Printf("Signalled done while waiting for htlc resolution, Correlationid: %d, Ignoring: %+v", index, h) + return s.defaultResolution() + case resolution := <-c: + log.Printf("Got resolution, Correlationid: %d: %+v", index, h) + return resolution + } +} + +func (s *server) recv() *HtlcResolution { + for { + sb := s.subscription + if sb == nil { + log.Printf("Got no subscribers for receive. Waiting for subscriber.") + select { + case <-s.done: + log.Printf("Done signalled, stopping receive.") + return s.defaultResolution() + case <-s.newSubscriber: + log.Printf("New subscription available for receive, continue receive.") + continue + } + } + + r, err := sb.stream.Recv() + if err == nil { + log.Printf("Received HtlcResolution %+v", r) + return r + } + + // TODO: close the subscription?? + log.Printf("Recv() errored, waiting %v: %v", receiveWaitDelay, err) + <-time.After(receiveWaitDelay) + } +} + +func (s *server) listenHtlcResponses() { + for { + select { + case <-s.done: + log.Printf("listenHtlcResponses received done. Stopping listening.") + return + case err := <-s.err: + log.Printf("listenHtlcResponses received error %v. Stopping listening.", err) + return + default: + response := s.recv() + s.corrMtx.Lock() + correlation, ok := s.correlations[response.Correlationid] + s.corrMtx.Unlock() + if ok { + correlation <- response + } else { + log.Printf("Got HTLC resolution that could not be correlated: %+v", response) + } + } + } +} + +func (s *server) defaultResolution() *HtlcResolution { + return &HtlcResolution{ + Outcome: &HtlcResolution_Continue{ + Continue: &HtlcContinue{}, + }, + } +}