add relay self-connection via authed pubkey
Some checks failed
Go / build-and-release (push) Has been cancelled
Some checks failed
Go / build-and-release (push) Has been cancelled
This commit is contained in:
5
.idea/.gitignore
generated
vendored
5
.idea/.gitignore
generated
vendored
@@ -1,5 +0,0 @@
|
|||||||
# Default ignored files
|
|
||||||
/shelf/
|
|
||||||
/workspace.xml
|
|
||||||
# Editor-based HTTP Client requests
|
|
||||||
/httpRequests/
|
|
||||||
8
.idea/modules.xml
generated
8
.idea/modules.xml
generated
@@ -1,8 +0,0 @@
|
|||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<project version="4">
|
|
||||||
<component name="ProjectModuleManager">
|
|
||||||
<modules>
|
|
||||||
<module fileurl="file://$PROJECT_DIR$/.idea/next.orly.dev.iml" filepath="$PROJECT_DIR$/.idea/next.orly.dev.iml" />
|
|
||||||
</modules>
|
|
||||||
</component>
|
|
||||||
</project>
|
|
||||||
6
.idea/vcs.xml
generated
6
.idea/vcs.xml
generated
@@ -1,6 +0,0 @@
|
|||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<project version="4">
|
|
||||||
<component name="VcsDirectoryMappings">
|
|
||||||
<mapping directory="" vcs="Git" />
|
|
||||||
</component>
|
|
||||||
</project>
|
|
||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
"lol.mleku.dev/log"
|
"lol.mleku.dev/log"
|
||||||
"next.orly.dev/pkg/database"
|
"next.orly.dev/pkg/database"
|
||||||
"next.orly.dev/pkg/database/indexes/types"
|
"next.orly.dev/pkg/database/indexes/types"
|
||||||
|
"git.mleku.dev/mleku/nostr/crypto/keys"
|
||||||
"git.mleku.dev/mleku/nostr/encoders/event"
|
"git.mleku.dev/mleku/nostr/encoders/event"
|
||||||
"git.mleku.dev/mleku/nostr/encoders/hex"
|
"git.mleku.dev/mleku/nostr/encoders/hex"
|
||||||
"git.mleku.dev/mleku/nostr/encoders/kind"
|
"git.mleku.dev/mleku/nostr/encoders/kind"
|
||||||
@@ -23,6 +24,7 @@ type ClusterManager struct {
|
|||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
db *database.D
|
db *database.D
|
||||||
adminNpubs []string
|
adminNpubs []string
|
||||||
|
relayIdentityPubkey string // Our relay's identity pubkey (hex)
|
||||||
members map[string]*ClusterMember // keyed by relay URL
|
members map[string]*ClusterMember // keyed by relay URL
|
||||||
membersMux sync.RWMutex
|
membersMux sync.RWMutex
|
||||||
pollTicker *time.Ticker
|
pollTicker *time.Ticker
|
||||||
@@ -30,6 +32,7 @@ type ClusterManager struct {
|
|||||||
httpClient *http.Client
|
httpClient *http.Client
|
||||||
propagatePrivilegedEvents bool
|
propagatePrivilegedEvents bool
|
||||||
publisher interface{ Deliver(*event.E) }
|
publisher interface{ Deliver(*event.E) }
|
||||||
|
nip11Cache *NIP11Cache
|
||||||
}
|
}
|
||||||
|
|
||||||
type ClusterMember struct {
|
type ClusterMember struct {
|
||||||
@@ -61,11 +64,20 @@ type EventInfo struct {
|
|||||||
func NewClusterManager(ctx context.Context, db *database.D, adminNpubs []string, propagatePrivilegedEvents bool, publisher interface{ Deliver(*event.E) }) *ClusterManager {
|
func NewClusterManager(ctx context.Context, db *database.D, adminNpubs []string, propagatePrivilegedEvents bool, publisher interface{ Deliver(*event.E) }) *ClusterManager {
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
|
|
||||||
|
// Get our relay identity pubkey
|
||||||
|
var relayPubkey string
|
||||||
|
if skb, err := db.GetRelayIdentitySecret(); err == nil && len(skb) == 32 {
|
||||||
|
if pk, err := keys.SecretBytesToPubKeyHex(skb); err == nil {
|
||||||
|
relayPubkey = pk
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
cm := &ClusterManager{
|
cm := &ClusterManager{
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
db: db,
|
db: db,
|
||||||
adminNpubs: adminNpubs,
|
adminNpubs: adminNpubs,
|
||||||
|
relayIdentityPubkey: relayPubkey,
|
||||||
members: make(map[string]*ClusterMember),
|
members: make(map[string]*ClusterMember),
|
||||||
pollDone: make(chan struct{}),
|
pollDone: make(chan struct{}),
|
||||||
propagatePrivilegedEvents: propagatePrivilegedEvents,
|
propagatePrivilegedEvents: propagatePrivilegedEvents,
|
||||||
@@ -73,6 +85,7 @@ func NewClusterManager(ctx context.Context, db *database.D, adminNpubs []string,
|
|||||||
httpClient: &http.Client{
|
httpClient: &http.Client{
|
||||||
Timeout: 30 * time.Second,
|
Timeout: 30 * time.Second,
|
||||||
},
|
},
|
||||||
|
nip11Cache: NewNIP11Cache(30 * time.Minute),
|
||||||
}
|
}
|
||||||
|
|
||||||
return cm
|
return cm
|
||||||
@@ -254,6 +267,12 @@ func (cm *ClusterManager) UpdateMembership(relayURLs []string) {
|
|||||||
|
|
||||||
// Add new members
|
// Add new members
|
||||||
for _, url := range relayURLs {
|
for _, url := range relayURLs {
|
||||||
|
// Skip if this is our own relay (check via NIP-11 pubkey)
|
||||||
|
if cm.isSelfRelay(url) {
|
||||||
|
log.D.F("skipping cluster member (self): %s (pubkey matches our relay identity)", url)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
if _, exists := cm.members[url]; !exists {
|
if _, exists := cm.members[url]; !exists {
|
||||||
// For simplicity, assume HTTP and WebSocket URLs are the same
|
// For simplicity, assume HTTP and WebSocket URLs are the same
|
||||||
// In practice, you'd need to parse these properly
|
// In practice, you'd need to parse these properly
|
||||||
@@ -269,6 +288,25 @@ func (cm *ClusterManager) UpdateMembership(relayURLs []string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isSelfRelay checks if a relay URL is actually ourselves by comparing NIP-11 pubkeys
|
||||||
|
func (cm *ClusterManager) isSelfRelay(relayURL string) bool {
|
||||||
|
// If we don't have a relay identity pubkey, can't compare
|
||||||
|
if cm.relayIdentityPubkey == "" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
peerPubkey, err := cm.nip11Cache.GetPubkey(ctx, relayURL)
|
||||||
|
if err != nil {
|
||||||
|
log.D.F("couldn't fetch NIP-11 for %s to check if self: %v", relayURL, err)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
return peerPubkey == cm.relayIdentityPubkey
|
||||||
|
}
|
||||||
|
|
||||||
// HandleMembershipEvent processes a cluster membership event (Kind 39108)
|
// HandleMembershipEvent processes a cluster membership event (Kind 39108)
|
||||||
func (cm *ClusterManager) HandleMembershipEvent(event *event.E) error {
|
func (cm *ClusterManager) HandleMembershipEvent(event *event.E) error {
|
||||||
// Verify the event is signed by a cluster admin
|
// Verify the event is signed by a cluster admin
|
||||||
@@ -313,6 +351,21 @@ func (cm *ClusterManager) HandleLatestSerial(w http.ResponseWriter, r *http.Requ
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Check if request is from ourselves by examining the Referer or Origin header
|
||||||
|
origin := r.Header.Get("Origin")
|
||||||
|
referer := r.Header.Get("Referer")
|
||||||
|
|
||||||
|
if origin != "" && cm.isSelfRelay(origin) {
|
||||||
|
log.D.F("rejecting cluster latest request from self (origin: %s)", origin)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if referer != "" && cm.isSelfRelay(referer) {
|
||||||
|
log.D.F("rejecting cluster latest request from self (referer: %s)", referer)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Get the latest serial from database by querying for the highest serial
|
// Get the latest serial from database by querying for the highest serial
|
||||||
latestSerial, err := cm.getLatestSerialFromDB()
|
latestSerial, err := cm.getLatestSerialFromDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -336,6 +389,21 @@ func (cm *ClusterManager) HandleEventsRange(w http.ResponseWriter, r *http.Reque
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Check if request is from ourselves by examining the Referer or Origin header
|
||||||
|
origin := r.Header.Get("Origin")
|
||||||
|
referer := r.Header.Get("Referer")
|
||||||
|
|
||||||
|
if origin != "" && cm.isSelfRelay(origin) {
|
||||||
|
log.D.F("rejecting cluster events request from self (origin: %s)", origin)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if referer != "" && cm.isSelfRelay(referer) {
|
||||||
|
log.D.F("rejecting cluster events request from self (referer: %s)", referer)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Parse query parameters
|
// Parse query parameters
|
||||||
fromStr := r.URL.Query().Get("from")
|
fromStr := r.URL.Query().Get("from")
|
||||||
toStr := r.URL.Query().Get("to")
|
toStr := r.URL.Query().Get("to")
|
||||||
|
|||||||
@@ -173,12 +173,36 @@ func (m *Manager) syncRoutine() {
|
|||||||
// syncWithPeersSequentially syncs with all configured peers one at a time
|
// syncWithPeersSequentially syncs with all configured peers one at a time
|
||||||
func (m *Manager) syncWithPeersSequentially() {
|
func (m *Manager) syncWithPeersSequentially() {
|
||||||
for _, peerURL := range m.peers {
|
for _, peerURL := range m.peers {
|
||||||
|
// Check if this peer is ourselves via NIP-11 pubkey
|
||||||
|
if m.isSelfPeer(peerURL) {
|
||||||
|
log.D.F("skipping sync with self: %s (pubkey matches our relay identity)", peerURL)
|
||||||
|
continue
|
||||||
|
}
|
||||||
m.syncWithPeer(peerURL)
|
m.syncWithPeer(peerURL)
|
||||||
// Small delay between peers to avoid overwhelming
|
// Small delay between peers to avoid overwhelming
|
||||||
time.Sleep(100 * time.Millisecond)
|
time.Sleep(100 * time.Millisecond)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isSelfPeer checks if a peer URL is actually ourselves by comparing NIP-11 pubkeys
|
||||||
|
func (m *Manager) isSelfPeer(peerURL string) bool {
|
||||||
|
// If we don't have a nodeID, can't compare
|
||||||
|
if m.nodeID == "" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
peerPubkey, err := m.nip11Cache.GetPubkey(ctx, peerURL)
|
||||||
|
if err != nil {
|
||||||
|
log.D.F("couldn't fetch NIP-11 for %s to check if self: %v", peerURL, err)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
return peerPubkey == m.nodeID
|
||||||
|
}
|
||||||
|
|
||||||
// syncWithPeer syncs with a specific peer
|
// syncWithPeer syncs with a specific peer
|
||||||
func (m *Manager) syncWithPeer(peerURL string) {
|
func (m *Manager) syncWithPeer(peerURL string) {
|
||||||
// Get the peer's current serial
|
// Get the peer's current serial
|
||||||
@@ -390,6 +414,13 @@ func (m *Manager) HandleCurrentRequest(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reject requests from ourselves (same nodeID)
|
||||||
|
if req.NodeID != "" && req.NodeID == m.nodeID {
|
||||||
|
log.D.F("rejecting sync current request from self (nodeID: %s)", req.NodeID)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
resp := CurrentResponse{
|
resp := CurrentResponse{
|
||||||
NodeID: m.nodeID,
|
NodeID: m.nodeID,
|
||||||
RelayURL: m.relayURL,
|
RelayURL: m.relayURL,
|
||||||
@@ -413,6 +444,13 @@ func (m *Manager) HandleEventIDsRequest(w http.ResponseWriter, r *http.Request)
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reject requests from ourselves (same nodeID)
|
||||||
|
if req.NodeID != "" && req.NodeID == m.nodeID {
|
||||||
|
log.D.F("rejecting sync event-ids request from self (nodeID: %s)", req.NodeID)
|
||||||
|
http.Error(w, "Cannot sync with self", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Get events with IDs in the requested range
|
// Get events with IDs in the requested range
|
||||||
eventMap, err := m.getEventsWithIDs(req.From, req.To)
|
eventMap, err := m.getEventsWithIDs(req.From, req.To)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -1 +1 @@
|
|||||||
v0.29.17
|
v0.29.18
|
||||||
Reference in New Issue
Block a user