diff --git a/events/event_stream_test.go b/events/event_stream_test.go index 599d839..93152c6 100644 --- a/events/event_stream_test.go +++ b/events/event_stream_test.go @@ -39,11 +39,7 @@ func (suite *EventStreamTestSuite) SetupTest() { } func (suite *EventStreamTestSuite) TestEventStart() { - evt := mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.1"), - } + evt := mtglib.NewEventStart("connID", net.ParseIP("10.0.0.1")) for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -52,10 +48,9 @@ func (suite *EventStreamTestSuite) TestEventStart() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventStart) - suite.Equal(evt.CreatedAt, caught.CreatedAt) - suite.Equal(evt.ConnID, caught.ConnID) suite.Equal(evt.RemoteIP.String(), caught.RemoteIP.String()) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) }) } @@ -64,12 +59,7 @@ func (suite *EventStreamTestSuite) TestEventStart() { } func (suite *EventStreamTestSuite) TestEventConnectedToDC() { - evt := mtglib.EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.1"), - DC: 3, - } + evt := mtglib.NewEventConnectedToDC("connID", net.ParseIP("10.0.0.1"), 3) for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -78,11 +68,10 @@ func (suite *EventStreamTestSuite) TestEventConnectedToDC() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventConnectedToDC) - suite.Equal(evt.CreatedAt, caught.CreatedAt) - suite.Equal(evt.ConnID, caught.ConnID) suite.Equal(evt.RemoteIP.String(), caught.RemoteIP.String()) suite.Equal(evt.StreamID(), caught.StreamID()) suite.Equal(evt.DC, caught.DC) + suite.Equal(evt.Timestamp(), caught.Timestamp()) }) } @@ -91,10 +80,7 @@ func (suite *EventStreamTestSuite) TestEventConnectedToDC() { } func (suite *EventStreamTestSuite) TestEventDomainFronting() { - evt := mtglib.EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: "connID", - } + evt := mtglib.NewEventDomainFronting("connID") for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -103,9 +89,8 @@ func (suite *EventStreamTestSuite) TestEventDomainFronting() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventDomainFronting) - suite.Equal(evt.CreatedAt, caught.CreatedAt) - suite.Equal(evt.ConnID, caught.ConnID) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) }) } @@ -114,12 +99,7 @@ func (suite *EventStreamTestSuite) TestEventDomainFronting() { } func (suite *EventStreamTestSuite) TestEventTraffic() { - evt := mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 1024, - IsRead: true, - } + evt := mtglib.NewEventTraffic("connID", 1024, true) for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -128,9 +108,8 @@ func (suite *EventStreamTestSuite) TestEventTraffic() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventTraffic) - suite.Equal(evt.CreatedAt, caught.CreatedAt) - suite.Equal(evt.ConnID, caught.ConnID) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) suite.Equal(evt.Traffic, caught.Traffic) suite.Equal(evt.IsRead, caught.IsRead) }) @@ -141,10 +120,7 @@ func (suite *EventStreamTestSuite) TestEventTraffic() { } func (suite *EventStreamTestSuite) TestEventFinish() { - evt := mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - } + evt := mtglib.NewEventFinish("connID") for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -153,9 +129,8 @@ func (suite *EventStreamTestSuite) TestEventFinish() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventFinish) - suite.Equal(evt.CreatedAt, caught.CreatedAt) - suite.Equal(evt.ConnID, caught.ConnID) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) }) } @@ -164,9 +139,7 @@ func (suite *EventStreamTestSuite) TestEventFinish() { } func (suite *EventStreamTestSuite) TestEventConcurrencyLimited() { - evt := mtglib.EventConcurrencyLimited{ - CreatedAt: time.Now(), - } + evt := mtglib.NewEventConcurrencyLimited() for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -175,7 +148,8 @@ func (suite *EventStreamTestSuite) TestEventConcurrencyLimited() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventConcurrencyLimited) - suite.Equal(evt.CreatedAt, caught.CreatedAt) + suite.Equal(evt.Timestamp(), caught.Timestamp()) + suite.Empty(evt.StreamID()) }) } @@ -184,10 +158,7 @@ func (suite *EventStreamTestSuite) TestEventConcurrencyLimited() { } func (suite *EventStreamTestSuite) TestEventIPBlocklisted() { - evt := mtglib.EventIPBlocklisted{ - CreatedAt: time.Now(), - RemoteIP: net.ParseIP("10.0.0.10"), - } + evt := mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10")) for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -196,8 +167,8 @@ func (suite *EventStreamTestSuite) TestEventIPBlocklisted() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventIPBlocklisted) - suite.Equal(evt.CreatedAt, caught.CreatedAt) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) suite.Equal(evt.RemoteIP.String(), caught.RemoteIP.String()) }) } @@ -207,10 +178,7 @@ func (suite *EventStreamTestSuite) TestEventIPBlocklisted() { } func (suite *EventStreamTestSuite) TestEventReplayAttack() { - evt := mtglib.EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: "CONNID", - } + evt := mtglib.NewEventReplayAttack("CONNID") for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { v. @@ -219,8 +187,8 @@ func (suite *EventStreamTestSuite) TestEventReplayAttack() { Run(func(args mock.Arguments) { caught := args.Get(0).(mtglib.EventReplayAttack) - suite.Equal(evt.CreatedAt, caught.CreatedAt) suite.Equal(evt.StreamID(), caught.StreamID()) + suite.Equal(evt.Timestamp(), caught.Timestamp()) }) } diff --git a/events/noop_test.go b/events/noop_test.go index 4a7b20b..369a061 100644 --- a/events/noop_test.go +++ b/events/noop_test.go @@ -4,7 +4,6 @@ import ( "context" "net" "testing" - "time" "github.com/9seconds/mtg/v2/events" "github.com/9seconds/mtg/v2/mtglib" @@ -20,42 +19,14 @@ type NoopTestSuite struct { func (suite *NoopTestSuite) SetupSuite() { suite.testData = map[string]mtglib.Event{ - "start": mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("127.0.0.1"), - }, - "connected-to-dc": mtglib.EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("127.1.0.1"), - DC: 2, - }, - "domain-fronting": mtglib.EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: "connID", - }, - "traffic": mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 1000, - IsRead: true, - }, - "finish": mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - }, - "concurrency-limited": mtglib.EventConcurrencyLimited{ - CreatedAt: time.Now(), - }, - "ip-blacklisted": mtglib.EventIPBlocklisted{ - RemoteIP: net.ParseIP("10.0.0.10"), - CreatedAt: time.Now(), - }, - "replay-attack": mtglib.EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: "connID", - }, + "start": mtglib.NewEventStart("connID", net.ParseIP("127.0.0.1")), + "connected-to-dc": mtglib.NewEventConnectedToDC("connID", net.ParseIP("127.1.0.1"), 2), + "domain-fronting": mtglib.NewEventDomainFronting("connID"), + "traffic": mtglib.NewEventTraffic("connID", 1000, true), + "finish": mtglib.NewEventFinish("connID"), + "concurrency-limited": mtglib.NewEventConcurrencyLimited(), + "ip-blacklisted": mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10")), + "replay-attack": mtglib.NewEventReplayAttack("connID"), } suite.ctx = context.Background() } diff --git a/mtglib/conns.go b/mtglib/conns.go index e6a64e7..da0b046 100644 --- a/mtglib/conns.go +++ b/mtglib/conns.go @@ -6,27 +6,21 @@ import ( "io" "net" "sync" - "time" ) type connTraffic struct { net.Conn - connID string - stream EventStream - ctx context.Context + streamID string + stream EventStream + ctx context.Context } func (c connTraffic) Read(b []byte) (int, error) { n, err := c.Conn.Read(b) if n > 0 { - c.stream.Send(c.ctx, EventTraffic{ - CreatedAt: time.Now(), - ConnID: c.connID, - Traffic: uint(n), - IsRead: true, - }) + c.stream.Send(c.ctx, NewEventTraffic(c.streamID, uint(n), true)) } return n, err // nolint: wrapcheck @@ -36,12 +30,7 @@ func (c connTraffic) Write(b []byte) (int, error) { n, err := c.Conn.Write(b) if n > 0 { - c.stream.Send(c.ctx, EventTraffic{ - CreatedAt: time.Now(), - ConnID: c.connID, - Traffic: uint(n), - IsRead: false, - }) + c.stream.Send(c.ctx, NewEventTraffic(c.streamID, uint(n), false)) } return n, err // nolint: wrapcheck diff --git a/mtglib/conns_internal_test.go b/mtglib/conns_internal_test.go index de31f60..0c73759 100644 --- a/mtglib/conns_internal_test.go +++ b/mtglib/conns_internal_test.go @@ -37,10 +37,10 @@ func (suite *ConnTrafficTestSuite) SetupTest() { suite.eventStreamMock = &EventStreamMock{} suite.connMock = &testlib.NetConnMock{} suite.conn = connTraffic{ - Conn: suite.connMock, - connID: "CONNID", - ctx: context.Background(), - stream: suite.eventStreamMock, + Conn: suite.connMock, + streamID: "CONNID", + ctx: context.Background(), + stream: suite.eventStreamMock, } } diff --git a/mtglib/events.go b/mtglib/events.go index e4c57d1..77c4c76 100644 --- a/mtglib/events.go +++ b/mtglib/events.go @@ -5,110 +5,133 @@ import ( "time" ) +type eventBase struct { + streamID string + timestamp time.Time +} + +func (e eventBase) StreamID() string { + return e.streamID +} + +func (e eventBase) Timestamp() time.Time { + return e.timestamp +} + type EventStart struct { - CreatedAt time.Time - ConnID string - RemoteIP net.IP -} + eventBase -func (e EventStart) StreamID() string { - return e.ConnID -} - -func (e EventStart) Timestamp() time.Time { - return e.CreatedAt + RemoteIP net.IP } type EventConnectedToDC struct { - CreatedAt time.Time - ConnID string - RemoteIP net.IP - DC int -} + eventBase -func (e EventConnectedToDC) StreamID() string { - return e.ConnID -} - -func (e EventConnectedToDC) Timestamp() time.Time { - return e.CreatedAt + RemoteIP net.IP + DC int } type EventTraffic struct { - CreatedAt time.Time - ConnID string - Traffic uint - IsRead bool -} + eventBase -func (e EventTraffic) StreamID() string { - return e.ConnID -} - -func (e EventTraffic) Timestamp() time.Time { - return e.CreatedAt + Traffic uint + IsRead bool } type EventFinish struct { - CreatedAt time.Time - ConnID string -} - -func (e EventFinish) StreamID() string { - return e.ConnID -} - -func (e EventFinish) Timestamp() time.Time { - return e.CreatedAt + eventBase } type EventDomainFronting struct { - CreatedAt time.Time - ConnID string -} - -func (e EventDomainFronting) StreamID() string { - return e.ConnID -} - -func (e EventDomainFronting) Timestamp() time.Time { - return e.CreatedAt + eventBase } type EventConcurrencyLimited struct { - CreatedAt time.Time -} - -func (e EventConcurrencyLimited) StreamID() string { - return "" -} - -func (e EventConcurrencyLimited) Timestamp() time.Time { - return e.CreatedAt + eventBase } type EventIPBlocklisted struct { - CreatedAt time.Time - RemoteIP net.IP -} + eventBase -func (e EventIPBlocklisted) StreamID() string { - return "" -} - -func (e EventIPBlocklisted) Timestamp() time.Time { - return e.CreatedAt + RemoteIP net.IP } type EventReplayAttack struct { - CreatedAt time.Time - ConnID string + eventBase } -func (e EventReplayAttack) StreamID() string { - return e.ConnID +func NewEventStart(streamID string, remoteIP net.IP) EventStart { + return EventStart{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + RemoteIP: remoteIP, + } } -func (e EventReplayAttack) Timestamp() time.Time { - return e.CreatedAt +func NewEventConnectedToDC(streamID string, remoteIP net.IP, dc int) EventConnectedToDC { + return EventConnectedToDC{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + RemoteIP: remoteIP, + DC: dc, + } +} + +func NewEventTraffic(streamID string, traffic uint, isRead bool) EventTraffic { + return EventTraffic{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + Traffic: traffic, + IsRead: isRead, + } +} + +func NewEventFinish(streamID string) EventFinish { + return EventFinish{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + } +} + +func NewEventDomainFronting(streamID string) EventDomainFronting { + return EventDomainFronting{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + } +} + +func NewEventConcurrencyLimited() EventConcurrencyLimited { + return EventConcurrencyLimited{ + eventBase: eventBase{ + timestamp: time.Now(), + }, + } +} + +func NewEventIPBlocklisted(remoteIP net.IP) EventIPBlocklisted { + return EventIPBlocklisted{ + eventBase: eventBase{ + timestamp: time.Now(), + }, + RemoteIP: remoteIP, + } +} + +func NewEventReplayAttack(streamID string) EventReplayAttack { + return EventReplayAttack{ + eventBase: eventBase{ + timestamp: time.Now(), + streamID: streamID, + }, + } } diff --git a/mtglib/events_test.go b/mtglib/events_test.go index 1f4814e..bb88b0a 100644 --- a/mtglib/events_test.go +++ b/mtglib/events_test.go @@ -14,84 +14,56 @@ type EventsTestSuite struct { } func (suite *EventsTestSuite) TestEventStart() { - evt := mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "CONNID", - RemoteIP: net.ParseIP("10.0.0.10"), - } + evt := mtglib.NewEventStart("CONNID", net.ParseIP("10.0.0.10")) suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventFinish() { - evt := mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "CONNID", - } + evt := mtglib.NewEventFinish("CONNID") suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventConnectedToDC() { - evt := mtglib.EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: "CONNID", - RemoteIP: net.ParseIP("10.0.0.10"), - DC: 3, - } + evt := mtglib.NewEventConnectedToDC("CONNID", net.ParseIP("10.0.0.10"), 3) suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventTraffic() { - evt := mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "CONNID", - Traffic: 3, - IsRead: true, - } + evt := mtglib.NewEventTraffic("CONNID", 1000, true) suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventDomainFronting() { - evt := mtglib.EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: "CONNID", - } + evt := mtglib.NewEventDomainFronting("CONNID") suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventConcurrencyLimited() { - evt := mtglib.EventConcurrencyLimited{ - CreatedAt: time.Now(), - } + evt := mtglib.NewEventConcurrencyLimited() suite.Empty(evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventIPBlocklisted() { - evt := mtglib.EventIPBlocklisted{ - CreatedAt: time.Now(), - RemoteIP: net.ParseIP("10.0.0.10"), - } + evt := mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10")) suite.Empty(evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } func (suite *EventsTestSuite) TestEventReplayAttack() { - evt := mtglib.EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: "CONNID", - } + evt := mtglib.NewEventReplayAttack("CONNID") suite.Equal("CONNID", evt.StreamID()) suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) diff --git a/mtglib/init.go b/mtglib/init.go index 9f3ba08..cd1b204 100644 --- a/mtglib/init.go +++ b/mtglib/init.go @@ -9,22 +9,66 @@ import ( ) var ( - ErrSecretEmpty = errors.New("secret is empty") - ErrSecretInvalid = errors.New("secret is invalid") - ErrNetworkIsNotDefined = errors.New("network is not defined") - ErrAntiReplayCacheIsNotDefined = errors.New("anti-replay cache is not defined") + // ErrSecretEmpty is returned if you are trying to create a proxy + // but do not provide a secret. + ErrSecretEmpty = errors.New("secret is empty") + + // ErrSecretInvalid is returned if you are trying to create a proxy + // but secret value is invalid (no host or payload are zeroes). + ErrSecretInvalid = errors.New("secret is invalid") + + // ErrNetworkIsNotDefined is returned if you are trying to create a + // proxy but network value is undefined. + ErrNetworkIsNotDefined = errors.New("network is not defined") + + // ErrAntiReplayCacheIsNotDefined is returned if you are trying to + // create a proxy but anti replay cache value is undefined. + ErrAntiReplayCacheIsNotDefined = errors.New("anti-replay cache is not defined") + + // ErrTimeAttackDetectorIsNotDefined is returned if you are trying to + // create a proxy but time attack detector is not defined. ErrTimeAttackDetectorIsNotDefined = errors.New("time attack detector is not defined") - ErrIPBlocklistIsNotDefined = errors.New("ip blocklist is not defined") - ErrEventStreamIsNotDefined = errors.New("event stream is not defined") - ErrLoggerIsNotDefined = errors.New("logger is not defined") + + // ErrIPBlocklistIsNotDefined is returned if you are trying to + // create a proxy but ip blocklist instance is not defined. + ErrIPBlocklistIsNotDefined = errors.New("ip blocklist is not defined") + + // ErrEventStreamIsNotDefined is returned if you are trying to create a + // proxy but event stream instance is not defined. + ErrEventStreamIsNotDefined = errors.New("event stream is not defined") + + // ErrLoggerIsNotDefined is returned if you are trying to + // create a proxy but logger is not defined. + ErrLoggerIsNotDefined = errors.New("logger is not defined") ) const ( - DefaultConcurrency = 4096 - DefaultBufferSize = 16 * 1024 // 16 kib + // DefaultConcurrency is a default max count of simultaneously + // connected clients. + DefaultConcurrency = 4096 + + // DefaultBufferSize is a default size of a copy buffer. + DefaultBufferSize = 16 * 1024 // 16 kib + + // DefaultDomainFrontingPort is a default port (HTTPS) to connect to in + // case of probe-resistance activity. DefaultDomainFrontingPort = 443 - DefaultIdleTimeout = time.Minute - DefaultPreferIP = "prefer-ipv6" + + // DefaultIdleTimeout is a default timeout for closing a connection + // in case of idling. + DefaultIdleTimeout = time.Minute + + // DefaultPreferIP is a default value for Telegram IP connectivity + // preference. + DefaultPreferIP = "prefer-ipv6" + + // SecretKeyLength defines a length of the secret bytes used + // by Telegram and a proxy. + SecretKeyLength = 16 + + // ConnectionIDBytesLength defines a count of random bytes + // used to generate a stream/connection ids. + ConnectionIDBytesLength = 16 ) // Network defines a knowledge how to work with a network. It may sound @@ -105,8 +149,24 @@ type IPBlocklist interface { Contains(net.IP) bool } +// Event is a data structure which is populated during mtg request +// processing lifecycle. Each request popluates many events: +// +// 1. Client connected +// +// 2. Request is finished +// +// 3. Connection to Telegram server is established +// +// and so on. All these events are data structures but all of them +// must conform the same interface. type Event interface { + // StreamID returns an identifier of the stream, connection, + // request, you name it. All events within the same stream returns + // the same stream id. StreamID() string + + // Timestamp returns a timestamp when this event was generated. Timestamp() time.Time } diff --git a/mtglib/proxy.go b/mtglib/proxy.go index 36de1a8..9400df5 100644 --- a/mtglib/proxy.go +++ b/mtglib/proxy.go @@ -53,18 +53,11 @@ func (p *Proxy) ServeConn(conn net.Conn) { ctx.Close() }() - p.eventStream.Send(ctx, EventStart{ - CreatedAt: time.Now(), - ConnID: ctx.connID, - RemoteIP: ctx.ClientIP(), - }) + p.eventStream.Send(ctx, NewEventStart(ctx.streamID, ctx.ClientIP())) ctx.logger.Info("Stream has been started") defer func() { - p.eventStream.Send(ctx, EventFinish{ - CreatedAt: time.Now(), - ConnID: ctx.connID, - }) + p.eventStream.Send(ctx, NewEventFinish(ctx.streamID)) ctx.logger.Info("Stream has been finished") }() @@ -109,10 +102,7 @@ func (p *Proxy) Serve(listener net.Listener) error { if p.ipBlocklist.Contains(ipAddr) { conn.Close() logger.Info("ip was blacklisted") - p.eventStream.Send(p.ctx, EventIPBlocklisted{ - CreatedAt: time.Now(), - RemoteIP: ipAddr, - }) + p.eventStream.Send(p.ctx, NewEventIPBlocklisted(ipAddr)) continue } @@ -125,9 +115,7 @@ func (p *Proxy) Serve(listener net.Listener) error { return nil case errors.Is(err, ants.ErrPoolOverload): logger.Info("connection was concurrency limited") - p.eventStream.Send(p.ctx, EventConcurrencyLimited{ - CreatedAt: time.Now(), - }) + p.eventStream.Send(p.ctx, NewEventConcurrencyLimited()) } select { @@ -181,10 +169,7 @@ func (p *Proxy) doFakeTLSHandshake(ctx *streamContext) bool { if p.antiReplayCache.SeenBefore(hello.SessionID) { p.logger.Warning("replay attack has been detected!") - p.eventStream.Send(p.ctx, EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: ctx.connID, - }) + p.eventStream.Send(p.ctx, NewEventReplayAttack(ctx.streamID)) p.doDomainFronting(ctx, rewind) return false @@ -235,31 +220,23 @@ func (p *Proxy) doTelegramCall(ctx *streamContext) error { ctx.telegramConn = obfuscated2.Conn{ Conn: connTraffic{ - Conn: conn, - connID: ctx.connID, - stream: p.eventStream, - ctx: ctx, + Conn: conn, + streamID: ctx.streamID, + stream: p.eventStream, + ctx: ctx, }, Encryptor: encryptor, Decryptor: decryptor, } - p.eventStream.Send(ctx, EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: ctx.connID, - RemoteIP: conn.RemoteAddr().(*net.TCPAddr).IP, - DC: ctx.dc, - }) + p.eventStream.Send(ctx, + NewEventConnectedToDC(ctx.streamID, conn.RemoteAddr().(*net.TCPAddr).IP, ctx.dc)) return nil } func (p *Proxy) doDomainFronting(ctx *streamContext, conn *connRewind) { - p.eventStream.Send(p.ctx, EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: ctx.connID, - }) - + p.eventStream.Send(p.ctx, NewEventDomainFronting(ctx.streamID)) conn.Rewind() frontConn, err := p.network.DialContext(ctx, "tcp", p.DomainFrontingAddress()) @@ -270,10 +247,10 @@ func (p *Proxy) doDomainFronting(ctx *streamContext, conn *connRewind) { } frontConn = connTraffic{ - Conn: frontConn, - ctx: ctx, - connID: ctx.connID, - stream: p.eventStream, + Conn: frontConn, + ctx: ctx, + streamID: ctx.streamID, + stream: p.eventStream, } rel := relay.AcquireRelay(ctx, diff --git a/mtglib/secret.go b/mtglib/secret.go index 71c4710..c76eed2 100644 --- a/mtglib/secret.go +++ b/mtglib/secret.go @@ -7,11 +7,7 @@ import ( "fmt" ) -const ( - SecretKeyLength = 16 - - secretFakeTLSFirstByte byte = 0xee -) +const secretFakeTLSFirstByte byte = 0xee var secretEmptyKey [SecretKeyLength]byte diff --git a/mtglib/stream_context.go b/mtglib/stream_context.go index 9b1027d..f704f37 100644 --- a/mtglib/stream_context.go +++ b/mtglib/stream_context.go @@ -8,14 +8,12 @@ import ( "time" ) -const ConnectionIDBytesLength = 16 - type streamContext struct { ctx context.Context ctxCancel context.CancelFunc clientConn net.Conn telegramConn net.Conn - connID string + streamID string dc int logger Logger } @@ -64,10 +62,10 @@ func newStreamContext(ctx context.Context, logger Logger, clientConn net.Conn) * ctx: ctx, ctxCancel: cancel, clientConn: clientConn, - connID: base64.RawURLEncoding.EncodeToString(connIDBytes), + streamID: base64.RawURLEncoding.EncodeToString(connIDBytes), } streamCtx.logger = logger. - BindStr("stream-id", streamCtx.connID). + BindStr("stream-id", streamCtx.streamID). BindStr("client-ip", streamCtx.ClientIP().String()) return streamCtx diff --git a/stats/prometheus_test.go b/stats/prometheus_test.go index abb3bec..6e44c0b 100644 --- a/stats/prometheus_test.go +++ b/stats/prometheus_test.go @@ -55,57 +55,39 @@ func (suite *PrometheusTestSuite) TearDownTest() { } func (suite *PrometheusTestSuite) TestTelegramPath() { - suite.prometheus.EventStart(mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.10"), - }) + suite.prometheus.EventStart( + mtglib.NewEventStart("connID", net.ParseIP("10.0.0.10"))) time.Sleep(100 * time.Millisecond) data, err := suite.Get() suite.NoError(err) suite.Contains(data, `mtg_client_connections{ip_family="ipv4"} 1`) - suite.prometheus.EventConnectedToDC(mtglib.EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.1"), - DC: 4, - }) + suite.prometheus.EventConnectedToDC( + mtglib.NewEventConnectedToDC("connID", net.ParseIP("10.0.0.1"), 4)) time.Sleep(100 * time.Millisecond) data, err = suite.Get() suite.NoError(err) suite.Contains(data, `mtg_telegram_connections{dc="4",telegram_ip="10.0.0.1"} 1`) - suite.prometheus.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 200, - IsRead: true, - }) + suite.prometheus.EventTraffic( + mtglib.NewEventTraffic("connID", 200, true)) time.Sleep(100 * time.Millisecond) data, err = suite.Get() suite.NoError(err) suite.Contains(data, `mtg_telegram_traffic{dc="4",direction="to_client",telegram_ip="10.0.0.1"} 200`) - suite.prometheus.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 100, - IsRead: false, - }) + suite.prometheus.EventTraffic( + mtglib.NewEventTraffic("connID", 100, false)) time.Sleep(100 * time.Millisecond) data, err = suite.Get() suite.NoError(err) suite.Contains(data, `mtg_telegram_traffic{dc="4",direction="from_client",telegram_ip="10.0.0.1"} 100`) - suite.prometheus.EventFinish(mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.prometheus.EventFinish(mtglib.NewEventFinish("connID")) time.Sleep(100 * time.Millisecond) data, err = suite.Get() @@ -115,21 +97,15 @@ func (suite *PrometheusTestSuite) TestTelegramPath() { } func (suite *PrometheusTestSuite) TestDomainFrontingPath() { - suite.prometheus.EventStart(mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.10"), - }) + suite.prometheus.EventStart( + mtglib.NewEventStart("connID", net.ParseIP("10.0.0.10"))) time.Sleep(100 * time.Millisecond) data, err := suite.Get() suite.NoError(err) suite.Contains(data, `mtg_client_connections{ip_family="ipv4"} 1`) - suite.prometheus.EventDomainFronting(mtglib.EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.prometheus.EventDomainFronting(mtglib.NewEventDomainFronting("connID")) time.Sleep(100 * time.Millisecond) data, err = suite.Get() @@ -137,34 +113,23 @@ func (suite *PrometheusTestSuite) TestDomainFrontingPath() { suite.Contains(data, `mtg_domain_fronting 1`) suite.Contains(data, `mtg_domain_fronting_connections{ip_family="ipv4"} 1`) - suite.prometheus.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 200, - IsRead: true, - }) + suite.prometheus.EventTraffic( + mtglib.NewEventTraffic("connID", 200, true)) time.Sleep(100 * time.Millisecond) data, err = suite.Get() suite.NoError(err) suite.Contains(data, `mtg_domain_fronting_traffic{direction="to_client"} 200`) - suite.prometheus.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 100, - IsRead: false, - }) + suite.prometheus.EventTraffic( + mtglib.NewEventTraffic("connID", 100, false)) time.Sleep(100 * time.Millisecond) data, err = suite.Get() suite.NoError(err) suite.Contains(data, `mtg_domain_fronting_traffic{direction="from_client"} 100`) - suite.prometheus.EventFinish(mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.prometheus.EventFinish(mtglib.NewEventFinish("connID")) time.Sleep(100 * time.Millisecond) data, err = suite.Get() @@ -174,9 +139,7 @@ func (suite *PrometheusTestSuite) TestDomainFrontingPath() { } func (suite *PrometheusTestSuite) TestEventConcurrencyLimited() { - suite.prometheus.EventConcurrencyLimited(mtglib.EventConcurrencyLimited{ - CreatedAt: time.Now(), - }) + suite.prometheus.EventConcurrencyLimited(mtglib.NewEventConcurrencyLimited()) time.Sleep(100 * time.Millisecond) @@ -186,10 +149,8 @@ func (suite *PrometheusTestSuite) TestEventConcurrencyLimited() { } func (suite *PrometheusTestSuite) TestEventIPBlocklisted() { - suite.prometheus.EventIPBlocklisted(mtglib.EventIPBlocklisted{ - CreatedAt: time.Now(), - RemoteIP: net.ParseIP("2001:db8::68"), - }) + suite.prometheus.EventIPBlocklisted( + mtglib.NewEventIPBlocklisted(net.ParseIP("2001:db8::68"))) time.Sleep(100 * time.Millisecond) @@ -199,10 +160,7 @@ func (suite *PrometheusTestSuite) TestEventIPBlocklisted() { } func (suite *PrometheusTestSuite) TestEventReplayAttack() { - suite.prometheus.EventReplayAttack(mtglib.EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.prometheus.EventReplayAttack(mtglib.NewEventReplayAttack("connID")) time.Sleep(100 * time.Millisecond) diff --git a/stats/statsd.go b/stats/statsd.go index 63bc005..f394ec6 100644 --- a/stats/statsd.go +++ b/stats/statsd.go @@ -4,7 +4,6 @@ import ( "fmt" "strconv" "strings" - "time" "github.com/9seconds/mtg/v2/events" "github.com/9seconds/mtg/v2/logger" @@ -123,14 +122,10 @@ func (s statsdProcessor) EventReplayAttack(_ mtglib.EventReplayAttack) { } func (s statsdProcessor) Shutdown() { - now := time.Now() events := make([]mtglib.EventFinish, 0, len(s.streams)) for k := range s.streams { - events = append(events, mtglib.EventFinish{ - CreatedAt: now, - ConnID: k, - }) + events = append(events, mtglib.NewEventFinish(k)) } for i := range events { diff --git a/stats/statsd_test.go b/stats/statsd_test.go index 02b34ec..4383509 100644 --- a/stats/statsd_test.go +++ b/stats/statsd_test.go @@ -105,48 +105,30 @@ func (suite *StatsdTestSuite) TearDownTest() { } func (suite *StatsdTestSuite) TestTelegramPath() { - suite.statsd.EventStart(mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.10"), - }) + suite.statsd.EventStart( + mtglib.NewEventStart("connID", net.ParseIP("10.0.0.10"))) time.Sleep(statsdSleepTime) suite.Equal("mtg.client_connections:+1|g|#ip_family:ipv4", suite.statsdServer.String()) - suite.statsd.EventConnectedToDC(mtglib.EventConnectedToDC{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.1.0.10"), - DC: 2, - }) + suite.statsd.EventConnectedToDC( + mtglib.NewEventConnectedToDC("connID", net.ParseIP("10.1.0.10"), 2)) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.telegram_connections:+1|g|#telegram_ip:10.1.0.10,dc:2") - suite.statsd.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 30, - IsRead: true, - }) + suite.statsd.EventTraffic( + mtglib.NewEventTraffic("connID", 30, true)) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.telegram_traffic:30|c|#telegram_ip:10.1.0.10,dc:2,direction:to_client") - suite.statsd.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 90, - IsRead: false, - }) + suite.statsd.EventTraffic( + mtglib.NewEventTraffic("connID", 90, false)) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.telegram_traffic:90|c|#telegram_ip:10.1.0.10,dc:2,direction:from_client") - suite.statsd.EventFinish(mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.statsd.EventFinish(mtglib.NewEventFinish("connID")) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.telegram_connections:-1|g|#telegram_ip:10.1.0.10,dc:2") @@ -158,47 +140,30 @@ func (suite *StatsdTestSuite) TestTelegramPath() { } func (suite *StatsdTestSuite) TestDomainFrontingPath() { - suite.statsd.EventStart(mtglib.EventStart{ - CreatedAt: time.Now(), - ConnID: "connID", - RemoteIP: net.ParseIP("10.0.0.10"), - }) + suite.statsd.EventStart( + mtglib.NewEventStart("connID", net.ParseIP("10.0.0.10"))) time.Sleep(statsdSleepTime) suite.Equal("mtg.client_connections:+1|g|#ip_family:ipv4", suite.statsdServer.String()) - suite.statsd.EventDomainFronting(mtglib.EventDomainFronting{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.statsd.EventDomainFronting(mtglib.NewEventDomainFronting("connID")) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.domain_fronting:1|c") suite.Contains(suite.statsdServer.String(), `mtg.domain_fronting_connections:+1|g|#ip_family:ipv4`) - suite.statsd.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 30, - IsRead: true, - }) + suite.statsd.EventTraffic( + mtglib.NewEventTraffic("connID", 30, true)) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), `mtg.domain_fronting_traffic:30|c|#direction:to_client`) - suite.statsd.EventTraffic(mtglib.EventTraffic{ - CreatedAt: time.Now(), - ConnID: "connID", - Traffic: 90, - IsRead: false, - }) + suite.statsd.EventTraffic( + mtglib.NewEventTraffic("connID", 90, false)) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), `mtg.domain_fronting_traffic:90|c|#direction:from_client`) - suite.statsd.EventFinish(mtglib.EventFinish{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.statsd.EventFinish(mtglib.NewEventFinish("connID")) time.Sleep(statsdSleepTime) suite.Contains(suite.statsdServer.String(), "mtg.domain_fronting_connections:-1|g|#ip_family:ipv4") @@ -210,29 +175,22 @@ func (suite *StatsdTestSuite) TestDomainFrontingPath() { } func (suite *StatsdTestSuite) TestEventConcurrencyLimited() { - suite.statsd.EventConcurrencyLimited(mtglib.EventConcurrencyLimited{ - CreatedAt: time.Now(), - }) + suite.statsd.EventConcurrencyLimited(mtglib.NewEventConcurrencyLimited()) time.Sleep(statsdSleepTime) suite.Equal("mtg.concurrency_limited:1|c", suite.statsdServer.String()) } func (suite *StatsdTestSuite) TestEventIPBlocklisted() { - suite.statsd.EventIPBlocklisted(mtglib.EventIPBlocklisted{ - CreatedAt: time.Now(), - RemoteIP: net.ParseIP("10.0.0.10"), - }) + suite.statsd.EventIPBlocklisted( + mtglib.NewEventIPBlocklisted(net.ParseIP("10.0.0.10"))) time.Sleep(statsdSleepTime) suite.Equal("mtg.ip_blocklisted:1|c", suite.statsdServer.String()) } func (suite *StatsdTestSuite) TestEventReplayAttack() { - suite.statsd.EventReplayAttack(mtglib.EventReplayAttack{ - CreatedAt: time.Now(), - ConnID: "connID", - }) + suite.statsd.EventReplayAttack(mtglib.NewEventReplayAttack("connID")) time.Sleep(statsdSleepTime) suite.Equal("mtg.replay_attacks:1|c", suite.statsdServer.String())