server.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743
  1. package oscar
  2. import (
  3. "bytes"
  4. "context"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "log/slog"
  9. "net"
  10. "net/netip"
  11. "sync"
  12. "time"
  13. "github.com/google/uuid"
  14. "github.com/patrickmn/go-cache"
  15. "golang.org/x/time/rate"
  16. "github.com/mk6i/open-oscar-server/config"
  17. "github.com/mk6i/open-oscar-server/server/oscar/middleware"
  18. "github.com/mk6i/open-oscar-server/state"
  19. "github.com/mk6i/open-oscar-server/wire"
  20. )
  21. func NewServer(
  22. authService AuthService,
  23. buddyListRegistry BuddyListRegistry,
  24. chatSessionManager *state.InMemoryChatSessionManager,
  25. departureNotifier DepartureNotifier,
  26. logger *slog.Logger,
  27. onlineNotifier OnlineNotifier,
  28. SNACHandler func(ctx context.Context, serverType uint16, instance *state.SessionInstance, inFrame wire.SNACFrame, r io.Reader, rw ResponseWriter, listener config.Listener) error,
  29. rateLimitUpdater RateLimitUpdater,
  30. limits wire.SNACRateLimits,
  31. limiter *IPRateLimiter,
  32. listenerCfg []config.Listener,
  33. recalcWarning func(ctx context.Context, instance *state.SessionInstance) error,
  34. lowerWarnLevel func(ctx context.Context, instance *state.SessionInstance),
  35. ) *Server {
  36. oscarSvc := oscarServer{
  37. authService: authService,
  38. buddyListRegistry: buddyListRegistry,
  39. chatSessionManager: chatSessionManager,
  40. departureNotifier: departureNotifier,
  41. logger: logger,
  42. onlineNotifier: onlineNotifier,
  43. snacHandler: SNACHandler,
  44. rateLimitUpdater: rateLimitUpdater,
  45. rateLimits: limits,
  46. ipRateLimiter: limiter,
  47. recalcWarning: recalcWarning,
  48. lowerWarnLevel: lowerWarnLevel,
  49. }
  50. ctx, cancel := context.WithCancel(context.Background())
  51. return &Server{
  52. closed: make(chan struct{}),
  53. conns: make(map[net.Conn]struct{}),
  54. handler: oscarSvc.routeConnection,
  55. listenerCfg: listenerCfg,
  56. logger: logger,
  57. shutdownCancel: cancel,
  58. shutdownCtx: ctx,
  59. }
  60. }
  61. type Server struct {
  62. logger *slog.Logger
  63. listenerCfg []config.Listener
  64. listeners []net.Listener
  65. connMu sync.Mutex
  66. conns map[net.Conn]struct{}
  67. connWg sync.WaitGroup
  68. listenWg sync.WaitGroup
  69. shutdownCtx context.Context
  70. shutdownCancel context.CancelFunc
  71. closed chan struct{}
  72. handler func(ctx context.Context, conn net.Conn, listener config.Listener) error
  73. }
  74. func (s *Server) ListenAndServe() error {
  75. for _, listenCfg := range s.listenerCfg {
  76. ln, err := net.Listen("tcp", listenCfg.BOSListenAddress)
  77. if err != nil {
  78. s.cleanupListeners()
  79. s.shutdownCancel()
  80. return fmt.Errorf("failed to listen on %s: %w", listenCfg.BOSListenAddress, err)
  81. }
  82. args := []any{
  83. "listen_address", listenCfg.BOSListenAddress,
  84. "advertised_host_plain", listenCfg.BOSAdvertisedHostPlain,
  85. }
  86. if listenCfg.HasSSL {
  87. args = append(args, "advertised_host_ssl", listenCfg.BOSAdvertisedHostSSL)
  88. }
  89. s.logger.Info("starting server", args...)
  90. s.listeners = append(s.listeners, ln)
  91. s.listenWg.Add(1)
  92. go s.acceptLoop(ln, listenCfg)
  93. }
  94. <-s.closed // block until Shutdown is called
  95. return nil
  96. }
  97. func (s *Server) Shutdown(ctx context.Context) error {
  98. s.logger.Debug("Initiating graceful shutdown...")
  99. s.shutdownCancel()
  100. s.cleanupListeners()
  101. // Wait for handlers to complete
  102. done := make(chan struct{})
  103. go func() {
  104. s.connWg.Wait()
  105. s.listenWg.Wait()
  106. close(done)
  107. }()
  108. select {
  109. case <-done:
  110. s.logger.Info("shutdown complete")
  111. case <-ctx.Done():
  112. s.logger.Info("shutdown complete, but connections didn't close cleanly")
  113. }
  114. close(s.closed)
  115. return nil
  116. }
  117. func (s *Server) acceptLoop(ln net.Listener, listener config.Listener) {
  118. defer s.listenWg.Done()
  119. for {
  120. conn, err := ln.Accept()
  121. if err != nil {
  122. if errors.Is(err, net.ErrClosed) {
  123. return
  124. }
  125. s.logger.Error("accept error", "err", err.Error())
  126. continue
  127. }
  128. // track connection
  129. s.connMu.Lock()
  130. s.conns[conn] = struct{}{}
  131. s.connMu.Unlock()
  132. s.connWg.Add(1)
  133. go s.handleConnection(s.shutdownCtx, conn, listener)
  134. }
  135. }
  136. func (s *Server) handleConnection(ctx context.Context, conn net.Conn, listener config.Listener) {
  137. defer func() {
  138. // untrack connections
  139. s.connMu.Lock()
  140. delete(s.conns, conn)
  141. s.connMu.Unlock()
  142. _ = conn.Close()
  143. s.connWg.Done()
  144. }()
  145. ctx = middleware.WithIP(ctx, conn.RemoteAddr().String())
  146. if err := s.handler(ctx, conn, listener); err != nil {
  147. s.logger.InfoContext(ctx, "user session failed", "err", err.Error())
  148. }
  149. }
  150. func (s *Server) cleanupListeners() {
  151. for _, ln := range s.listeners {
  152. _ = ln.Close()
  153. }
  154. s.listeners = nil
  155. }
  156. type oscarServer struct {
  157. authService AuthService
  158. buddyListRegistry BuddyListRegistry
  159. chatSessionManager ChatSessionManager
  160. departureNotifier DepartureNotifier
  161. logger *slog.Logger
  162. onlineNotifier OnlineNotifier
  163. snacHandler func(ctx context.Context, serverType uint16, instance *state.SessionInstance, inFrame wire.SNACFrame, r io.Reader, rw ResponseWriter, listener config.Listener) error
  164. rateLimitUpdater RateLimitUpdater
  165. rateLimits wire.SNACRateLimits
  166. ipRateLimiter *IPRateLimiter
  167. recalcWarning func(ctx context.Context, instance *state.SessionInstance) error
  168. lowerWarnLevel func(ctx context.Context, instance *state.SessionInstance)
  169. }
  170. func (s oscarServer) routeConnection(ctx context.Context, conn net.Conn, listener config.Listener) error {
  171. ip, _, err := net.SplitHostPort(conn.RemoteAddr().String())
  172. if err != nil {
  173. s.logger.Error("failed to parse remote address", "err", err.Error())
  174. return err
  175. }
  176. flapc := wire.NewFlapClient(100, conn, conn)
  177. // send flap signon with server capabilities
  178. if err := flapc.SendSignonFrame(nil); err != nil {
  179. return err
  180. }
  181. flap, err := flapc.ReceiveSignonFrame()
  182. if err != nil {
  183. return err
  184. }
  185. if flap.HasTag(wire.OServiceTLVTagsLoginCookie) {
  186. return s.connectToOSCARService(ctx, flap, flapc, conn, listener)
  187. }
  188. return s.authenticate(ctx, flap, ip, conn, flapc, listener.BOSAdvertisedHostPlain)
  189. }
  190. func (s oscarServer) connectToOSCARService(
  191. ctx context.Context,
  192. flap wire.FLAPSignonFrame,
  193. flapc *wire.FlapClient,
  194. conn net.Conn,
  195. listener config.Listener,
  196. ) error {
  197. authCookie, ok := flap.Bytes(wire.OServiceTLVTagsLoginCookie)
  198. if !ok {
  199. return errors.New("unable to get session id from payload")
  200. }
  201. cookie, err := s.authService.CrackCookie(authCookie)
  202. if err != nil {
  203. return err
  204. }
  205. s.logger.Debug("connecting to service", "service", wire.FoodGroupName(cookie.Service))
  206. var instance *state.SessionInstance
  207. switch cookie.Service {
  208. case wire.BOS:
  209. sessCfg := func(sess *state.Session) {
  210. sess.OnSessionClose(func() {
  211. if !shuttingDown(ctx) {
  212. if err := s.departureNotifier.BroadcastBuddyDeparted(ctx, sess.IdentScreenName()); err != nil {
  213. s.logger.ErrorContext(ctx, "error sending buddy departure notifications", "err", err.Error())
  214. }
  215. }
  216. ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
  217. defer cancel()
  218. // buddy list must be cleared before session is closed, otherwise
  219. // there will be a race condition that could cause the buddy list
  220. // be prematurely deleted.
  221. if err := s.buddyListRegistry.UnregisterBuddyList(ctx, instance.IdentScreenName()); err != nil {
  222. s.logger.ErrorContext(ctx, "error removing buddy list entry", "err", err.Error())
  223. }
  224. s.chatSessionManager.RemoveUserFromAllChats(instance.IdentScreenName())
  225. s.authService.Signout(ctx, sess)
  226. })
  227. }
  228. instance, err = s.authService.RegisterBOSSession(ctx, cookie, sessCfg)
  229. if err != nil {
  230. if errors.Is(err, state.ErrMaxConcurrentSessionsReached) {
  231. s.logger.Debug("session registration failed", "err", err.Error())
  232. block := wire.TLVRestBlock{}
  233. // error code indicating the signon is blocked. i can't find a
  234. // more appropriate error code to indicate the maximum session limit is reached
  235. block.Append(wire.NewTLVBE(0x0008, uint8(0x18)))
  236. if err := flapc.NewSignoff(block); err != nil {
  237. return fmt.Errorf("unable to gracefully disconnect user. %w", err)
  238. }
  239. return nil
  240. }
  241. return err
  242. }
  243. if instance == nil {
  244. return errors.New("session not found")
  245. }
  246. defer func() {
  247. instance.CloseInstance()
  248. }()
  249. if err = instance.Session().RunOnce(func() error {
  250. // make buddy list visible to other users
  251. if err := s.buddyListRegistry.RegisterBuddyList(ctx, instance.IdentScreenName()); err != nil {
  252. return fmt.Errorf("unable to init buddy list: %w", err)
  253. }
  254. // restore warning level from last session
  255. if err := s.recalcWarning(ctx, instance); err != nil {
  256. return fmt.Errorf("failed to recalculate warning level: %w", err)
  257. }
  258. // periodically decay warning level
  259. go s.lowerWarnLevel(ctx, instance)
  260. return nil
  261. }); err != nil {
  262. return err
  263. }
  264. // Update user visibility when an instance closes, as the user's overall status may change.
  265. // Example: With 1 away and 1 non-away instance, the user appears available. If the non-away
  266. // instance closes, the user should appear away.
  267. instance.OnClose(func() {
  268. if shuttingDown(ctx) {
  269. return
  270. }
  271. if instance.Session().Invisible() {
  272. if err := s.departureNotifier.BroadcastBuddyDeparted(ctx, instance.IdentScreenName()); err != nil {
  273. s.logger.ErrorContext(ctx, "error sending buddy departure notifications", "err", err.Error())
  274. }
  275. } else {
  276. if err := s.departureNotifier.BroadcastBuddyArrived(ctx, instance.IdentScreenName(), instance.Session().TLVUserInfo()); err != nil {
  277. s.logger.ErrorContext(ctx, "error sending buddy arrival notifications", "err", err.Error())
  278. }
  279. }
  280. })
  281. if remoteAddr, ok := middleware.IPFromContext(ctx); ok {
  282. ip, err := netip.ParseAddrPort(remoteAddr)
  283. if err != nil {
  284. return errors.New("unable to parse ip addr")
  285. }
  286. instance.SetRemoteAddr(&ip)
  287. }
  288. go s.receiveSessMessages(ctx, instance, flapc)
  289. case wire.Chat:
  290. sessCfg := func(sess *state.Session) {
  291. sess.OnSessionClose(func() {
  292. ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
  293. defer cancel()
  294. s.authService.SignoutChat(ctx, sess)
  295. })
  296. }
  297. instance, err = s.authService.RegisterChatSession(ctx, cookie, sessCfg)
  298. if err != nil {
  299. return err
  300. }
  301. if instance == nil {
  302. return errors.New("session not found")
  303. }
  304. defer func() {
  305. instance.CloseInstance()
  306. }()
  307. go s.receiveSessMessages(ctx, instance, flapc)
  308. default:
  309. instance, err = s.authService.RetrieveBOSSession(ctx, cookie)
  310. if err != nil {
  311. return err
  312. }
  313. if instance == nil {
  314. return errors.New("session not found")
  315. }
  316. }
  317. ctx = middleware.WithScreenName(ctx, instance.IdentScreenName())
  318. msg := s.onlineNotifier.HostOnline(cookie.Service)
  319. if err := flapc.SendSNAC(msg.Frame, msg.Body); err != nil {
  320. return err
  321. }
  322. return s.dispatchIncomingMessages(ctx, cookie.Service, instance, flapc, conn, listener)
  323. }
  324. func shuttingDown(ctx context.Context) bool {
  325. select {
  326. case <-ctx.Done():
  327. // server is shutting down, don't send buddy notifications
  328. return true
  329. default:
  330. }
  331. return false
  332. }
  333. func (s oscarServer) receiveSessMessages(ctx context.Context, instance *state.SessionInstance, flapc *wire.FlapClient) {
  334. for {
  335. select {
  336. case <-instance.Closed():
  337. return
  338. case <-ctx.Done():
  339. return
  340. case m := <-instance.ReceiveMessage():
  341. // forward a notification sent from another client to this client
  342. if err := flapc.SendSNAC(m.Frame, m.Body); err != nil {
  343. middleware.LogRequestError(ctx, s.logger, m.Frame, err)
  344. } else {
  345. middleware.LogRequest(ctx, s.logger, m.Frame, m.Body)
  346. }
  347. }
  348. }
  349. }
  350. func (s oscarServer) authenticate(
  351. ctx context.Context,
  352. flap wire.FLAPSignonFrame,
  353. ip string,
  354. conn net.Conn,
  355. flapc *wire.FlapClient,
  356. advertisedHost string,
  357. ) error {
  358. if ok, isBUCP := s.ipRateLimiter.Allow(ip); !ok {
  359. s.logger.InfoContext(ctx, "user rate limited at login, dropping connection")
  360. tlv := wire.TLVRestBlock{
  361. TLVList: []wire.TLV{
  362. wire.NewTLVBE(wire.LoginTLVTagsErrorSubcode, wire.LoginErrRateLimitExceeded),
  363. },
  364. }
  365. // gives wrong response if you quickly switch between BUCP/FLAP clients
  366. if isBUCP {
  367. return flapc.SendSNAC(
  368. wire.SNACFrame{
  369. FoodGroup: wire.BUCP,
  370. SubGroup: wire.BUCPLoginResponse,
  371. },
  372. wire.SNAC_0x17_0x03_BUCPLoginResponse{
  373. TLVRestBlock: tlv,
  374. },
  375. )
  376. } else {
  377. return flapc.NewSignoff(tlv)
  378. }
  379. }
  380. // auth must complete within the next 30 seconds
  381. if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil {
  382. return fmt.Errorf("failed to set deadline: %w", err)
  383. }
  384. // decide whether the client is using BUCP or FLAP authentication based on
  385. // the presence of the screen name TLV. this block used to check for the
  386. // presence of the roasted password TLV, however that proved an unreliable
  387. // indicator of FLAP-auth because older ICQ clients appear to omit the
  388. // roasted password TLV when the password is not stored client-side.
  389. if _, hasScreenName := flap.Uint16BE(wire.LoginTLVTagsScreenName); hasScreenName {
  390. return s.processFLAPAuth(ctx, flap, flapc, advertisedHost)
  391. }
  392. s.ipRateLimiter.SetBUCP(ip)
  393. return s.processBUCPAuth(ctx, flapc, advertisedHost)
  394. }
  395. func (s oscarServer) processFLAPAuth(
  396. ctx context.Context,
  397. signonFrame wire.FLAPSignonFrame,
  398. flapc *wire.FlapClient,
  399. advertisedHost string,
  400. ) error {
  401. tlv, err := s.authService.FLAPLogin(ctx, signonFrame, advertisedHost)
  402. if err != nil {
  403. return err
  404. }
  405. return flapc.NewSignoff(tlv)
  406. }
  407. func (s oscarServer) processBUCPAuth(ctx context.Context, flapc *wire.FlapClient, advertisedHost string) error {
  408. frames := 0
  409. for {
  410. frame, err := flapc.ReceiveFLAP()
  411. if err != nil {
  412. return err
  413. }
  414. if frames > 10 {
  415. // a lot of frames received, the client is misbehaving
  416. return fmt.Errorf("too many auth flap packets received")
  417. }
  418. frames++
  419. switch frame.FrameType {
  420. case wire.FLAPFrameSignoff:
  421. s.logger.Debug("signed off mid-login")
  422. return io.EOF // client disconnected
  423. case wire.FLAPFrameKeepAlive:
  424. s.logger.Debug("received flap keepalive frame")
  425. case wire.FLAPFrameData:
  426. buf := bytes.NewReader(frame.Payload)
  427. fr := wire.SNACFrame{}
  428. if err := wire.UnmarshalBE(&fr, buf); err != nil {
  429. return err
  430. }
  431. switch {
  432. case fr.FoodGroup == wire.BUCP && fr.SubGroup == wire.BUCPChallengeRequest:
  433. challengeRequest := wire.SNAC_0x17_0x06_BUCPChallengeRequest{}
  434. if err := wire.UnmarshalBE(&challengeRequest, buf); err != nil {
  435. return err
  436. }
  437. outSNAC, err := s.authService.BUCPChallenge(ctx, challengeRequest, uuid.New)
  438. if err != nil {
  439. return err
  440. }
  441. outSNAC.Frame.RequestID = fr.RequestID
  442. if err := flapc.SendSNAC(outSNAC.Frame, outSNAC.Body); err != nil {
  443. return err
  444. }
  445. if outSNAC.Frame.SubGroup == wire.BUCPLoginResponse {
  446. screenName, _ := challengeRequest.String(wire.LoginTLVTagsScreenName)
  447. s.logger.Debug("failed BUCP challenge: user does not exist", "screen_name", screenName)
  448. return nil // account does not exist
  449. }
  450. case fr.FoodGroup == wire.BUCP && fr.SubGroup == wire.BUCPLoginRequest:
  451. loginRequest := wire.SNAC_0x17_0x02_BUCPLoginRequest{}
  452. if err := wire.UnmarshalBE(&loginRequest, buf); err != nil {
  453. return err
  454. }
  455. outSNAC, err := s.authService.BUCPLogin(ctx, loginRequest, advertisedHost)
  456. if err != nil {
  457. return err
  458. }
  459. outSNAC.Frame.RequestID = fr.RequestID
  460. // Clients expect login response as SNAC on FLAP
  461. // channel 2 followed by a FLAP signoff frame to properly close the auth
  462. // connection
  463. if err := flapc.SendSNAC(outSNAC.Frame, outSNAC.Body); err != nil {
  464. return err
  465. }
  466. return flapc.NewSignoff(wire.TLVRestBlock{})
  467. default:
  468. s.logger.Debug("unexpected SNAC received during login",
  469. "foodgroup", wire.FoodGroupName(fr.FoodGroup),
  470. "subgroup", wire.SubGroupName(fr.FoodGroup, fr.SubGroup))
  471. return io.EOF
  472. }
  473. default:
  474. s.logger.Debug("unexpected frame type received during login", "type", frame.FrameType)
  475. return io.EOF
  476. }
  477. }
  478. }
  479. func sendInvalidSNACErr(frameIn wire.SNACFrame, rw ResponseWriter) error {
  480. frameOut := wire.SNACFrame{
  481. FoodGroup: frameIn.FoodGroup,
  482. SubGroup: 0x01, // error subgroup for all SNACs
  483. RequestID: frameIn.RequestID,
  484. }
  485. bodyOut := wire.SNACError{
  486. Code: wire.ErrorCodeInvalidSnac,
  487. }
  488. return rw.SendSNAC(frameOut, bodyOut)
  489. }
  490. // dispatchIncomingMessages receives incoming messages and sends them to the
  491. // appropriate message handler. Messages from the client are sent to the
  492. // router. Messages relayed from the user session are forwarded to the client.
  493. // This function ensures that the same sequence number is incremented for both
  494. // types of messages. The function terminates upon receiving a connection error
  495. // or when the session closes.
  496. func (s oscarServer) dispatchIncomingMessages(
  497. ctx context.Context,
  498. fg uint16,
  499. instance *state.SessionInstance,
  500. flapc *wire.FlapClient,
  501. r io.ReadCloser,
  502. listener config.Listener,
  503. ) error {
  504. defer func() {
  505. s.logger.InfoContext(ctx, "user disconnected")
  506. }()
  507. // buffered so that the go routine has room to exit
  508. msgCh := make(chan wire.FLAPFrame, 1)
  509. errCh := make(chan error, 1)
  510. // consume flap frames
  511. go func() {
  512. defer close(msgCh)
  513. defer close(errCh)
  514. for {
  515. frame := wire.FLAPFrame{}
  516. if err := wire.UnmarshalBE(&frame, r); err != nil {
  517. errCh <- err
  518. return
  519. }
  520. msgCh <- frame
  521. }
  522. }()
  523. for {
  524. select {
  525. case flap, ok := <-msgCh:
  526. if !ok {
  527. return nil
  528. }
  529. switch flap.FrameType {
  530. case wire.FLAPFrameData:
  531. flapBuf := bytes.NewBuffer(flap.Payload)
  532. inFrame := wire.SNACFrame{}
  533. if err := wire.UnmarshalBE(&inFrame, flapBuf); err != nil {
  534. return err
  535. }
  536. rateClassID, ok := s.rateLimits.RateClassLookup(inFrame.FoodGroup, inFrame.SubGroup)
  537. if ok {
  538. if status := instance.Session().EvaluateRateLimit(time.Now(), rateClassID); status == wire.RateLimitStatusLimited {
  539. s.logger.DebugContext(ctx, "rate limit exceeded, dropping SNAC",
  540. "foodgroup", wire.FoodGroupName(inFrame.FoodGroup),
  541. "subgroup", wire.SubGroupName(inFrame.FoodGroup, inFrame.SubGroup))
  542. break
  543. }
  544. } else {
  545. s.logger.ErrorContext(ctx, "rate limit not found, allowing request through")
  546. }
  547. // route a client request to the appropriate service handler. the
  548. // handler may write a response to the client connection.
  549. if err := s.snacHandler(ctx, fg, instance, inFrame, flapBuf, flapc, listener); err != nil {
  550. middleware.LogRequestError(ctx, s.logger, inFrame, err)
  551. if errors.Is(err, ErrRouteNotFound) {
  552. if err1 := sendInvalidSNACErr(inFrame, flapc); err1 != nil {
  553. return errors.Join(err1, err)
  554. }
  555. break
  556. }
  557. return err
  558. }
  559. case wire.FLAPFrameSignon:
  560. return fmt.Errorf("shouldn't get FLAPFrameSignon. flap: %v", flap)
  561. case wire.FLAPFrameError:
  562. return fmt.Errorf("got FLAPFrameError. flap: %v", flap)
  563. case wire.FLAPFrameSignoff:
  564. s.logger.InfoContext(ctx, "got FLAPFrameSignoff", "flap", flap)
  565. return nil
  566. case wire.FLAPFrameKeepAlive:
  567. s.logger.DebugContext(ctx, "keepalive heartbeat")
  568. default:
  569. return fmt.Errorf("got unknown FLAP frame type. flap: %v", flap)
  570. }
  571. case <-time.After(1 * time.Second):
  572. updates := s.rateLimitUpdater.RateLimitUpdates(ctx, instance, time.Now())
  573. for _, update := range updates {
  574. if err := flapc.SendSNAC(update.Frame, update.Body); err != nil {
  575. middleware.LogRequestError(ctx, s.logger, update.Frame, err)
  576. return err
  577. }
  578. }
  579. case <-instance.Closed():
  580. // add logoff reason to clients that support multi-conn
  581. if instance.MultiConnFlag() == wire.MultiConnFlagsOldClient {
  582. if err := flapc.OldSignoff(); err != nil {
  583. return fmt.Errorf("unable to gracefully disconnect user. %w", err)
  584. }
  585. } else {
  586. block := wire.TLVRestBlock{}
  587. // error code indicating user signed in a different location
  588. block.Append(wire.NewTLVBE(0x0009, wire.OServiceDiscErrNewLogin))
  589. // "more info" button
  590. block.Append(wire.NewTLVBE(0x000b, "https://github.com/mk6i/open-oscar-server"))
  591. if err := flapc.NewSignoff(block); err != nil {
  592. return fmt.Errorf("unable to gracefully disconnect user. %w", err)
  593. }
  594. }
  595. return nil
  596. case <-ctx.Done():
  597. if instance.MultiConnFlag() == wire.MultiConnFlagsOldClient {
  598. if err := flapc.OldSignoff(); err != nil {
  599. return fmt.Errorf("unable to gracefully disconnect user. %w", err)
  600. }
  601. } else {
  602. if err := flapc.NewSignoff(wire.TLVRestBlock{}); err != nil {
  603. return fmt.Errorf("unable to gracefully disconnect user. %w", err)
  604. }
  605. }
  606. return nil
  607. case err := <-errCh:
  608. if !errors.Is(err, io.EOF) {
  609. s.logger.ErrorContext(ctx, "client disconnected with error", "err", err)
  610. }
  611. return nil
  612. }
  613. }
  614. }
  615. // IPRateLimiter enforces a per-IP rate limit using a token bucket algorithm.
  616. // It caches individual rate limiters by IP address and supports tagging requests
  617. // as originating from the BUCP or FLAP auth.
  618. //
  619. // The limiter uses an in-memory cache with TTL expiration, so rate limits reset
  620. // after the TTL if no activity is observed for a given IP.
  621. type IPRateLimiter struct {
  622. cache *cache.Cache // In-memory cache mapping IPs to rate limiters with optional BUCP tag
  623. rate rate.Limit // Requests allowed per second
  624. burst int // Maximum burst size allowed
  625. }
  626. type rateLimitEntry struct {
  627. isBUCP bool
  628. limiter *rate.Limiter
  629. }
  630. // NewIPRateLimiter initializes a new IPRateLimiter with the specified rate,
  631. // burst size, and TTL for each IP's limiter. Entries expire after 2×TTL.
  632. func NewIPRateLimiter(rate rate.Limit, burst int, ttl time.Duration) *IPRateLimiter {
  633. return &IPRateLimiter{
  634. cache: cache.New(ttl, 2*ttl),
  635. rate: rate,
  636. burst: burst,
  637. }
  638. }
  639. // SetBUCP marks the rate limiter for the given IP as originating from BUCP auth
  640. // (default FLAP auth).
  641. func (l *IPRateLimiter) SetBUCP(ip string) {
  642. limiter, found := l.cache.Get(ip)
  643. if !found {
  644. limiter = &rateLimitEntry{
  645. isBUCP: true,
  646. limiter: rate.NewLimiter(l.rate, l.burst),
  647. }
  648. l.cache.Set(ip, limiter, cache.DefaultExpiration)
  649. }
  650. limiter.(*rateLimitEntry).isBUCP = true
  651. }
  652. // Allow checks if a request from the given IP is allowed under its rate limit.
  653. // It returns whether the request is allowed and whether the connection uses
  654. // BUCP auth.
  655. func (l *IPRateLimiter) Allow(ip string) (allowed bool, isBUCP bool) {
  656. limiter, found := l.cache.Get(ip)
  657. if !found {
  658. limiter = &rateLimitEntry{
  659. limiter: rate.NewLimiter(l.rate, l.burst),
  660. }
  661. l.cache.Set(ip, limiter, cache.DefaultExpiration)
  662. }
  663. entry := limiter.(*rateLimitEntry)
  664. return entry.limiter.Allow(), entry.isBUCP
  665. }