mirror of
https://github.com/0ceanSlim/grain.git
synced 2024-11-23 17:07:13 +00:00
Compare commits
No commits in common. "8491827b0627662d07c47973ab2b3741bad732f6" and "4fef7088bb603cd76d632a5e327b6a4afbcd60e3" have entirely different histories.
8491827b06
...
4fef7088bb
@ -11,7 +11,6 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"grain/config"
|
"grain/config"
|
||||||
nostr "grain/server/types"
|
|
||||||
|
|
||||||
"golang.org/x/net/websocket"
|
"golang.org/x/net/websocket"
|
||||||
)
|
)
|
||||||
@ -54,20 +53,13 @@ func ImportEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
// Filter events based on the whitelist
|
err = sendEventsToRelay(events)
|
||||||
whitelistedEvents := filterWhitelistedEvents(events)
|
|
||||||
if len(whitelistedEvents) == 0 {
|
|
||||||
log.Printf("No whitelisted events to import from relay %s", url)
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
err = sendEventsToRelay(whitelistedEvents)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errorChan <- fmt.Errorf("error sending events to relay: %w", err)
|
errorChan <- fmt.Errorf("error sending events to relay: %w", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
totalEvents += len(whitelistedEvents)
|
totalEvents += len(events)
|
||||||
|
|
||||||
// Update lastEventCreatedAt with the timestamp of the last event fetched
|
// Update lastEventCreatedAt with the timestamp of the last event fetched
|
||||||
lastEventCreatedAt = int64(events[len(events)-1]["created_at"].(float64))
|
lastEventCreatedAt = int64(events[len(events)-1]["created_at"].(float64))
|
||||||
@ -87,40 +79,6 @@ func ImportEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// filterWhitelistedEvents filters events based on the whitelist configuration.
|
|
||||||
func filterWhitelistedEvents(events []map[string]interface{}) []map[string]interface{} {
|
|
||||||
var whitelistedEvents []map[string]interface{}
|
|
||||||
|
|
||||||
// Load the whitelist configuration
|
|
||||||
whitelistCfg := config.GetWhitelistConfig()
|
|
||||||
if whitelistCfg == nil {
|
|
||||||
log.Println("Whitelist configuration is not loaded. Allowing all events.")
|
|
||||||
return events
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, event := range events {
|
|
||||||
evt := nostr.Event{
|
|
||||||
ID: event["id"].(string),
|
|
||||||
PubKey: event["pubkey"].(string),
|
|
||||||
CreatedAt: int64(event["created_at"].(float64)),
|
|
||||||
Kind: int(event["kind"].(float64)),
|
|
||||||
Content: event["content"].(string),
|
|
||||||
Tags: event["tags"].([][]string),
|
|
||||||
Sig: event["sig"].(string),
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check the whitelist criteria
|
|
||||||
isWhitelisted, _ := config.CheckWhitelist(evt)
|
|
||||||
if isWhitelisted {
|
|
||||||
whitelistedEvents = append(whitelistedEvents, event)
|
|
||||||
} else {
|
|
||||||
log.Printf("Event ID %s blocked due to whitelist rules", evt.ID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return whitelistedEvents
|
|
||||||
}
|
|
||||||
|
|
||||||
func renderResult(w http.ResponseWriter, success bool, message string, count int) {
|
func renderResult(w http.ResponseWriter, success bool, message string, count int) {
|
||||||
tmpl, err := template.New("result").Parse(`
|
tmpl, err := template.New("result").Parse(`
|
||||||
{{ if .Success }}
|
{{ if .Success }}
|
||||||
|
@ -14,10 +14,6 @@ server:
|
|||||||
max_connections: 100
|
max_connections: 100
|
||||||
max_subscriptions_per_client: 10
|
max_subscriptions_per_client: 10
|
||||||
|
|
||||||
backup_relay:
|
|
||||||
enabled: false # Set to true to enable sending events to the backup relay
|
|
||||||
url: "wss://some-relay.com" # URL of the backup relay
|
|
||||||
|
|
||||||
event_time_constraints:
|
event_time_constraints:
|
||||||
min_created_at: 1577836800 # January 1, 2020, as Unix timestamp
|
min_created_at: 1577836800 # January 1, 2020, as Unix timestamp
|
||||||
# min_created_at_string: now-5m # Custom value to indicate 5 minutes in the past
|
# min_created_at_string: now-5m # Custom value to indicate 5 minutes in the past
|
||||||
|
@ -26,8 +26,4 @@ type ServerConfig struct {
|
|||||||
Auth AuthConfig `yaml:"auth"`
|
Auth AuthConfig `yaml:"auth"`
|
||||||
EventPurge EventPurgeConfig `yaml:"event_purge"`
|
EventPurge EventPurgeConfig `yaml:"event_purge"`
|
||||||
EventTimeConstraints EventTimeConstraints `yaml:"event_time_constraints"` // Added this field
|
EventTimeConstraints EventTimeConstraints `yaml:"event_time_constraints"` // Added this field
|
||||||
BackupRelay struct {
|
|
||||||
Enabled bool `yaml:"enabled"`
|
|
||||||
URL string `yaml:"url"`
|
|
||||||
} `yaml:"backup_relay"`
|
|
||||||
}
|
}
|
||||||
|
@ -1,25 +0,0 @@
|
|||||||
package mongo
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
nostr "grain/server/types" // Adjust import path as needed
|
|
||||||
|
|
||||||
"go.mongodb.org/mongo-driver/bson"
|
|
||||||
)
|
|
||||||
|
|
||||||
// CheckDuplicateEvent checks if an event with the same ID already exists in the collection.
|
|
||||||
func CheckDuplicateEvent(ctx context.Context, evt nostr.Event) (bool, error) {
|
|
||||||
collection := GetCollection(evt.Kind)
|
|
||||||
filter := bson.M{"id": evt.ID}
|
|
||||||
|
|
||||||
var existingEvent nostr.Event
|
|
||||||
err := collection.FindOne(ctx, filter).Decode(&existingEvent)
|
|
||||||
if err != nil {
|
|
||||||
if err.Error() == "mongo: no documents in result" {
|
|
||||||
return false, nil // No duplicate found
|
|
||||||
}
|
|
||||||
return false, fmt.Errorf("error checking for duplicate event: %v", err)
|
|
||||||
}
|
|
||||||
return true, nil // Duplicate found
|
|
||||||
}
|
|
@ -4,7 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"grain/config"
|
"grain/config"
|
||||||
types "grain/config/types"
|
types "grain/config/types"
|
||||||
nostr "grain/server/types"
|
"grain/server/utils"
|
||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@ -12,7 +12,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// PurgeOldEvents removes old events based on the configuration and a list of whitelisted pubkeys.
|
// PurgeOldEvents removes old events based on the configuration and a list of whitelisted pubkeys.
|
||||||
func PurgeOldEvents(cfg *types.EventPurgeConfig) {
|
func PurgeOldEvents(cfg *types.EventPurgeConfig, whitelist []string) {
|
||||||
if !cfg.Enabled {
|
if !cfg.Enabled {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -23,49 +23,78 @@ func PurgeOldEvents(cfg *types.EventPurgeConfig) {
|
|||||||
// Calculate the cutoff time
|
// Calculate the cutoff time
|
||||||
cutoff := time.Now().AddDate(0, 0, -cfg.KeepDurationDays).Unix()
|
cutoff := time.Now().AddDate(0, 0, -cfg.KeepDurationDays).Unix()
|
||||||
|
|
||||||
// Create the base filter for fetching old events
|
// Create the filter for purging old events
|
||||||
baseFilter := bson.M{
|
filter := bson.M{
|
||||||
"created_at": bson.M{"$lt": cutoff}, // Filter for events older than the cutoff
|
"created_at": bson.M{"$lt": cutoff}, // Filter older events
|
||||||
}
|
}
|
||||||
|
|
||||||
cursor, err := collection.Find(context.TODO(), baseFilter)
|
// Exclude whitelisted pubkeys if specified in the config
|
||||||
if err != nil {
|
if cfg.ExcludeWhitelisted && len(whitelist) > 0 {
|
||||||
log.Printf("Error fetching old events for purging: %v", err)
|
filter["pubkey"] = bson.M{"$nin": whitelist} // Exclude whitelisted pubkeys
|
||||||
return
|
|
||||||
}
|
}
|
||||||
defer cursor.Close(context.TODO())
|
|
||||||
|
|
||||||
for cursor.Next(context.TODO()) {
|
// Handle purging by category
|
||||||
var evt nostr.Event
|
for category, purge := range cfg.PurgeByCategory {
|
||||||
if err := cursor.Decode(&evt); err != nil {
|
if purge {
|
||||||
log.Printf("Error decoding event: %v", err)
|
filter["category"] = category
|
||||||
continue
|
_, err := collection.DeleteMany(context.TODO(), filter)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("Error purging events by category %s: %v", category, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Check if the event's pubkey is whitelisted and skip purging if configured to do so
|
// Handle purging by kind
|
||||||
if cfg.ExcludeWhitelisted && config.IsPubKeyWhitelisted(evt.PubKey) {
|
for _, kindRule := range cfg.PurgeByKind {
|
||||||
log.Printf("Skipping purging for whitelisted event ID: %s, pubkey: %s", evt.ID, evt.PubKey)
|
if kindRule.Enabled {
|
||||||
continue
|
filter["kind"] = kindRule.Kind
|
||||||
}
|
_, err := collection.DeleteMany(context.TODO(), filter)
|
||||||
|
if err != nil {
|
||||||
// Proceed with deleting the event if it is not whitelisted
|
log.Printf("Error purging events by kind %d: %v", kindRule.Kind, err)
|
||||||
_, err := collection.DeleteOne(context.TODO(), bson.M{"id": evt.ID})
|
}
|
||||||
if err != nil {
|
|
||||||
log.Printf("Error purging event ID %s: %v", evt.ID, err)
|
|
||||||
} else {
|
|
||||||
log.Printf("Purged event ID: %s", evt.ID)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ScheduleEventPurging runs the event purging at a configurable interval.
|
// ScheduleEventPurging runs the event purging at a configurable interval.
|
||||||
func ScheduleEventPurging(cfg *types.ServerConfig) {
|
func ScheduleEventPurging(cfg *types.ServerConfig) {
|
||||||
|
// Use the purge interval from the configuration
|
||||||
purgeInterval := time.Duration(cfg.EventPurge.PurgeIntervalHours) * time.Hour
|
purgeInterval := time.Duration(cfg.EventPurge.PurgeIntervalHours) * time.Hour
|
||||||
ticker := time.NewTicker(purgeInterval)
|
ticker := time.NewTicker(purgeInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
for range ticker.C {
|
for range ticker.C {
|
||||||
PurgeOldEvents(&cfg.EventPurge)
|
// Fetch the whitelisted pubkeys without passing cfg directly
|
||||||
log.Println("Scheduled purging completed.")
|
whitelist := getWhitelistedPubKeys()
|
||||||
|
PurgeOldEvents(&cfg.EventPurge, whitelist)
|
||||||
|
log.Printf("Purged old events, keeping whitelisted pubkeys: %v", whitelist)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Fetch whitelisted pubkeys from both the whitelist config and any additional domains.
|
||||||
|
func getWhitelistedPubKeys() []string {
|
||||||
|
// Get the whitelist configuration
|
||||||
|
whitelistCfg := config.GetWhitelistConfig()
|
||||||
|
if whitelistCfg == nil {
|
||||||
|
log.Println("whitelistCfg is nil, returning an empty list of whitelisted pubkeys.")
|
||||||
|
return []string{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Start with the statically defined pubkeys
|
||||||
|
whitelistedPubkeys := whitelistCfg.PubkeyWhitelist.Pubkeys
|
||||||
|
|
||||||
|
// Fetch pubkeys from domains if domain whitelist is enabled
|
||||||
|
if whitelistCfg.DomainWhitelist.Enabled {
|
||||||
|
domains := whitelistCfg.DomainWhitelist.Domains
|
||||||
|
pubkeys, err := utils.FetchPubkeysFromDomains(domains)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("Error fetching pubkeys from domains: %v", err)
|
||||||
|
// Return the existing statically whitelisted pubkeys in case of an error
|
||||||
|
return whitelistedPubkeys
|
||||||
|
}
|
||||||
|
// Append fetched pubkeys from domains to the whitelisted pubkeys
|
||||||
|
whitelistedPubkeys = append(whitelistedPubkeys, pubkeys...)
|
||||||
|
}
|
||||||
|
|
||||||
|
return whitelistedPubkeys
|
||||||
|
}
|
@ -6,7 +6,6 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"grain/config"
|
"grain/config"
|
||||||
"grain/server/db/mongo"
|
"grain/server/db/mongo"
|
||||||
"log"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"grain/server/handlers/response"
|
"grain/server/handlers/response"
|
||||||
@ -67,66 +66,9 @@ func HandleEvent(ws *websocket.Conn, message []interface{}) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check for duplicate event
|
|
||||||
isDuplicate, err := mongo.CheckDuplicateEvent(context.TODO(), evt)
|
|
||||||
if err != nil {
|
|
||||||
fmt.Printf("Error checking for duplicate event: %v\n", err)
|
|
||||||
response.SendOK(ws, evt.ID, false, "error: internal server error during duplicate check")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if isDuplicate {
|
|
||||||
response.SendOK(ws, evt.ID, false, "blocked: the database already contains this event")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Store the event in MongoDB or other storage
|
// Store the event in MongoDB or other storage
|
||||||
mongo.StoreMongoEvent(context.TODO(), evt, ws)
|
mongo.StoreMongoEvent(context.TODO(), evt, ws)
|
||||||
fmt.Println("Event processed:", evt.ID)
|
fmt.Println("Event processed:", evt.ID)
|
||||||
|
|
||||||
// Load the config and check for errors
|
|
||||||
cfg, err := config.LoadConfig("config.yml")
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("Error loading configuration: %v", err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Send the event to the backup relay if configured
|
|
||||||
if cfg.BackupRelay.Enabled {
|
|
||||||
go func() {
|
|
||||||
err := sendToBackupRelay(cfg.BackupRelay.URL, evt)
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("Failed to send event %s to backup relay: %v", evt.ID, err)
|
|
||||||
} else {
|
|
||||||
log.Printf("Event %s successfully sent to backup relay", evt.ID)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func sendToBackupRelay(backupURL string, evt nostr.Event) error {
|
|
||||||
conn, err := websocket.Dial(backupURL, "", "http://localhost/")
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("error connecting to backup relay %s: %w", backupURL, err)
|
|
||||||
}
|
|
||||||
defer conn.Close()
|
|
||||||
|
|
||||||
// Create the message to send
|
|
||||||
eventMessage := []interface{}{"EVENT", evt}
|
|
||||||
eventMessageBytes, err := json.Marshal(eventMessage)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("error marshaling event message: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := conn.Write(eventMessageBytes); err != nil {
|
|
||||||
return fmt.Errorf("error sending event message to backup relay: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Log and return
|
|
||||||
log.Printf("Event %s sent to backup relay %s", evt.ID, backupURL)
|
|
||||||
time.Sleep(500 * time.Millisecond) // Optional: small delay to avoid rapid successive sends
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Validate event timestamps against the configured min and max values
|
// Validate event timestamps against the configured min and max values
|
||||||
@ -164,20 +106,20 @@ func validateEventTimestamp(evt nostr.Event) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func handleBlacklistAndWhitelist(ws *websocket.Conn, evt nostr.Event) bool {
|
func handleBlacklistAndWhitelist(ws *websocket.Conn, evt nostr.Event) bool {
|
||||||
// Use the updated CheckBlacklist function
|
// Use the updated CheckBlacklist function
|
||||||
if blacklisted, msg := config.CheckBlacklist(evt.PubKey, evt.Content); blacklisted {
|
if blacklisted, msg := config.CheckBlacklist(evt.PubKey, evt.Content); blacklisted {
|
||||||
response.SendOK(ws, evt.ID, false, msg)
|
response.SendOK(ws, evt.ID, false, msg)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check the whitelist using CheckWhitelist function
|
// Check the whitelist using CheckWhitelist function
|
||||||
isWhitelisted, msg := config.CheckWhitelist(evt)
|
isWhitelisted, msg := config.CheckWhitelist(evt)
|
||||||
if !isWhitelisted {
|
if !isWhitelisted {
|
||||||
response.SendOK(ws, evt.ID, false, msg)
|
response.SendOK(ws, evt.ID, false, msg)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func handleRateAndSizeLimits(ws *websocket.Conn, evt nostr.Event, eventSize int) bool {
|
func handleRateAndSizeLimits(ws *websocket.Conn, evt nostr.Event, eventSize int) bool {
|
||||||
|
Loading…
Reference in New Issue
Block a user