wip
This commit is contained in:
Sebastien Blot
parent
ebfa952c28
commit
0df40e8f89
|
|
@ -1,61 +0,0 @@
|
|||
package syslogserver
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
|
||||
"gopkg.in/tomb.v2"
|
||||
)
|
||||
|
||||
type SyslogServer struct {
|
||||
proto string
|
||||
listenAddr string
|
||||
port int
|
||||
channel chan string
|
||||
tcpListener *net.TCPListener
|
||||
udpConn *net.UDPConn
|
||||
parsingTombs []*tomb.Tomb
|
||||
acceptTombs []*tomb.Tomb
|
||||
receiveTombs []*tomb.Tomb
|
||||
}
|
||||
|
||||
func (s *SyslogServer) SetProtocol(proto string) error {
|
||||
proto = strings.ToLower(proto)
|
||||
if proto != "tcp" && proto != "udp" {
|
||||
return fmt.Errorf("protocol must be tcp or udp, got %s", proto)
|
||||
}
|
||||
s.proto = proto
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogServer) Listen(listenAddr string, port int) error {
|
||||
switch s.proto {
|
||||
case "tcp":
|
||||
tcpAddr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", s.listenAddr, s.port))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tcpListener, err := net.ListenTCP("tcp", tcpAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.tcpListener = tcpListener
|
||||
case "udp":
|
||||
udpAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", s.listenAddr, s.port))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
udpConn, err := net.ListenUDP("tcp", udpAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.udpConn = udpConn
|
||||
s.udpConn.SetReadBuffer(1024 * 8) // FIXME probably
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogServer) SetChannel(c chan string) {
|
||||
s.channel = c
|
||||
}
|
||||
|
|
@ -1,140 +0,0 @@
|
|||
package syslog
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
|
||||
syslogserver "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/syslog/internal"
|
||||
"github.com/crowdsecurity/crowdsec/pkg/types"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"gopkg.in/tomb.v2"
|
||||
"gopkg.in/yaml.v2"
|
||||
)
|
||||
|
||||
type SyslogConfiguration struct {
|
||||
Proto string `yaml:"protocol,omitempty"`
|
||||
Port int `yaml:"port,omitempty"`
|
||||
Addr string `yaml:"addr,omitempty"`
|
||||
//TODO: Add TLS support
|
||||
configuration.DataSourceCommonCfg `yaml:",inline"`
|
||||
}
|
||||
|
||||
type SyslogSource struct {
|
||||
config SyslogConfiguration
|
||||
logger *log.Entry
|
||||
server *syslogserver.SyslogServer
|
||||
}
|
||||
|
||||
func (s *SyslogSource) GetName() string {
|
||||
return "syslog"
|
||||
}
|
||||
|
||||
func (s *SyslogSource) GetMode() string {
|
||||
return s.config.Mode
|
||||
}
|
||||
|
||||
func (s *SyslogSource) Dump() interface{} {
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *SyslogSource) CanRun() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogSource) GetMetrics() []prometheus.Collector {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogSource) ConfigureByDSN(dsn string, labelType string, logger *log.Entry) error {
|
||||
return fmt.Errorf("syslog datasource does not support one shot acquisition")
|
||||
}
|
||||
|
||||
func (s *SyslogSource) OneShotAcquisition(out chan types.Event, t *tomb.Tomb) error {
|
||||
return fmt.Errorf("syslog datasource does not support one shot acquisition")
|
||||
}
|
||||
|
||||
func (s *SyslogSource) Configure(yamlConfig []byte, logger *log.Entry) error {
|
||||
s.logger = logger
|
||||
s.logger.Infof("Starting syslog datasource configuration")
|
||||
syslogConfig := SyslogConfiguration{}
|
||||
syslogConfig.Mode = configuration.TAIL_MODE
|
||||
err := yaml.UnmarshalStrict(yamlConfig, &syslogConfig)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "Cannot parse syslog configuration")
|
||||
}
|
||||
if syslogConfig.Addr == "" {
|
||||
syslogConfig.Addr = "127.0.0.1" //do we want a usable or secure default ?
|
||||
}
|
||||
if syslogConfig.Port == 0 {
|
||||
syslogConfig.Port = 514
|
||||
}
|
||||
if syslogConfig.Proto == "" {
|
||||
syslogConfig.Proto = "udp"
|
||||
}
|
||||
s.config = syslogConfig
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) error {
|
||||
//channel := make(syslog.LogPartsChannel)
|
||||
//handler := syslog.NewChannelHandler(channel)
|
||||
|
||||
c := make(chan string)
|
||||
s.server = &syslogserver.SyslogServer{}
|
||||
err := s.server.SetProtocol(s.config.Proto)
|
||||
s.server.SetChannel(c)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "could not set syslog server protocol")
|
||||
}
|
||||
err = s.server.Listen(s.config.Addr, s.config.Port)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "could not start syslog server")
|
||||
}
|
||||
//s.server.SetHandler(handler)
|
||||
//err := s.server.ListenUDP(fmt.Sprintf("%s:%d", s.config.Addr, s.config.Port))
|
||||
/*if err != nil {
|
||||
return errors.Wrap(err, "could not listen")
|
||||
}
|
||||
err = s.server.Boot()
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "could not start syslog server")
|
||||
}*/
|
||||
t.Go(func() error {
|
||||
defer types.CatchPanic("crowdsec/acquis/syslog/live")
|
||||
return s.handleSyslogMsg(out, t, c)
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SyslogSource) handleSyslogMsg(out chan types.Event, t *tomb.Tomb, c chan string) error {
|
||||
for {
|
||||
select {
|
||||
case <-t.Dying():
|
||||
s.logger.Info("Syslog datasource is dying")
|
||||
case syslogLine := <-c:
|
||||
//var line string
|
||||
spew.Dump(syslogLine)
|
||||
//rebuild the syslog line from the part
|
||||
//TODO: handle the RFC format and cases such as missing PID, or PID embedded in the app_name
|
||||
/*if logParts["content"] == nil {
|
||||
line = fmt.Sprintf("%s %s %s[%s]: %s", logParts["timestamp"], logParts["hostname"],
|
||||
logParts["app_name"], logParts["proc_id"], logParts["message"])
|
||||
} else {
|
||||
line = fmt.Sprintf("%s %s %s: %s", logParts["timestamp"],
|
||||
logParts["hostname"], logParts["tag"], logParts["content"])
|
||||
}
|
||||
l := types.Line{}
|
||||
l.Raw = line
|
||||
l.Labels = s.config.Labels
|
||||
//l.Time = logParts["timestamp"].(string)
|
||||
l.Src = logParts["client"].(string)
|
||||
l.Process = true
|
||||
out <- types.Event{Line: l, Process: true, Type: types.LOG, ExpectMode: leaky.LIVE}*/
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue