updated messaging system
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Paul 2023-01-15 16:04:35 +01:00
parent 9244e05fc3
commit 4bada1c166
3 changed files with 30 additions and 21 deletions

View File

@ -16,23 +16,27 @@ func (event *Event) Insert(cfg *config.Config) (err error) {
return
}
/*
func (event *Event) APIFormat() *APIEvent {
if event == nil {
return nil
}
return &APIEvent{
IP: event.IP.IP,
Src: event.Src,
Hostname: event.Hostname.String,
MSgType: "none",
IPData: {
IP: event.IP.IP,
Src: event.Src,
Hostname: event.Hostname.String,
},
}
}
}*/
func (event *Event) APIParse(session *xorm.Session, apievent APIEvent) (err error) {
*event = Event{
IP: &IP{IP: apievent.IP},
Src: apievent.Src,
IP: &IP{IP: apievent.IPData.IP},
Src: apievent.IPData.Src,
Hostname: sql.NullString{
String: apievent.Hostname,
String: apievent.IPData.Hostname,
Valid: true},
}
event.IP.GetOrCreate(session)
@ -49,9 +53,12 @@ type Event struct {
}
type APIEvent struct {
IP string `json:"ip"`
Src string `json:"src"`
Hostname string `json:"hostname"`
Mode string `json:"mode"`
Created string `json:"created"`
MsgType string `json:"msgtype"`
IPData struct {
IP string `json:"ip"`
Src string `json:"src"`
Hostname string `json:"hostname"`
Mode string `json:"mode"`
Created string `json:"created"`
} `json:"ipdata"`
}

View File

@ -14,7 +14,9 @@ import (
"xorm.io/xorm"
)
const ScanLimit = 10
const SCANLIMIT = 10
const IPINFO_WS = "https://ip.paulbsd.com"
var lastday = time.Now().Add(-(time.Hour * 24))
@ -130,10 +132,11 @@ func InsertIPBulk(session *xorm.Session, ips *[]IP) (numinsert int64, numupdate
}
func ScanIP(cfg *config.Config) (err error) {
queryclient := http.Client{}
for {
session := cfg.Db.NewSession()
orphans := []IP{}
err = session.Where("(as_id is null or country_id is null or city_id is null or rdns = '') and updated < now()-'1d'::interval").Asc("updated").Limit(ScanLimit).Find(&orphans)
err = session.Where("(as_id is null or country_id is null or city_id is null or rdns = '') and updated < now()-'1d'::interval").Asc("updated").Limit(SCANLIMIT).Find(&orphans)
session.Close()
if err == nil && len(orphans) > 0 {
@ -142,7 +145,7 @@ func ScanIP(cfg *config.Config) (err error) {
log.Println(err)
}
for _, orphan := range orphans {
query, err := QueryInfo(orphan.IP)
query, err := QueryInfo(&queryclient, orphan.IP)
if err != nil {
log.Println(err)
continue
@ -181,9 +184,8 @@ func ScanIP(cfg *config.Config) (err error) {
}
}
func QueryInfo(ip string) (query QueryIP, err error) {
client := http.Client{}
var url = fmt.Sprintf("https://ip.paulbsd.com/%s", ip)
func QueryInfo(client *http.Client, ip string) (query QueryIP, err error) {
var url = fmt.Sprintf("%s/%s", IPINFO_WS, ip)
req, _ := http.NewRequest("GET", url, nil)
req.Header.Add("Accept", "*/*")
req.Header.Add("User-Agent", "ipbl")

View File

@ -51,11 +51,11 @@ func Handle(cfg *config.Config, reqsock *goczmq.Sock, pubsock *goczmq.Sock, chan
log.Println("unable to parse ip address", err)
continue
}
if apievent.IP != "" && apievent.IP == lastip {
if apievent.IPData.IP != "" && apievent.IPData.IP == lastip {
continue
}
if apievent.Mode != "init" {
if apievent.MsgType != "init" {
session := cfg.Db.NewSession()
event.APIParse(session, apievent)
session.Close()
@ -69,7 +69,7 @@ func Handle(cfg *config.Config, reqsock *goczmq.Sock, pubsock *goczmq.Sock, chan
tmpval := fmt.Sprintf("%s %s", channel, string(val))
ipjson := []byte(tmpval)
topub = append(topub, ipjson)
lastip = apievent.IP
lastip = apievent.IPData.IP
}
err = pubsock.SendMessage(topub)