cmd/mmc/main.go

   1 // mmc -- Mattermost client
   2 // Copyright (C) 2023-2024 Sergey Matveev <stargrave@stargrave.org>
   3 //
   4 // This program is free software: you can redistribute it and/or modify
   5 // it under the terms of the GNU Affero General Public License as
   6 // published by the Free Software Foundation, either version 3 of the
   7 // License.
   8 //
   9 // This program is distributed in the hope that it will be useful,
  10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 // GNU Affero General Public License for more details.
  13 //
  14 // You should have received a copy of the GNU Affero General Public License
  15 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 package main
  18
  19 import (
  20         "archive/tar"
  21         "bytes"
  22         "crypto/tls"
  23         "encoding/json"
  24         "errors"
  25         "flag"
  26         "fmt"
  27         "io"
  28         "io/fs"
  29         "log"
  30         "net/http"
  31         "net/url"
  32         "os"
  33         "os/exec"
  34         "os/signal"
  35         "path"
  36         "sort"
  37         "strings"
  38         "sync"
  39         "syscall"
  40         "time"
  41
  42         "github.com/davecgh/go-spew/spew"
  43         "github.com/gorilla/websocket"
  44         "github.com/mattermost/mattermost-server/v6/model"
  45         "go.cypherpunks.su/netrc/v2"
  46         "go.stargrave.org/mmc/internal"
  47 )
  48
  49 var (
  50         Newwin   = flag.String("newwin", "cmd/newwin", "Path to newwin command")
  51         DebugFd  *os.File
  52         UmaskCur int
  53 )
  54
  55 func rewriteIfChanged(fn string, data string) {
  56         if their, err := os.ReadFile(fn); err != nil ||
  57                 !bytes.Equal([]byte(data), their) {
  58                 if err = os.WriteFile(fn, []byte(data), 0o666); err != nil {
  59                         log.Fatalln(err)
  60                 }
  61         }
  62 }
  63
  64 func mkFifo(pth string) {
  65         if _, err := os.Stat(pth); err == nil {
  66                 return
  67         }
  68         if err := syscall.Mkfifo(pth, 0666); err != nil {
  69                 log.Fatalln(err)
  70         }
  71 }
  72
  73 func main() {
  74         entrypoint := flag.String("entrypoint", mmc.GetEntrypoint(), "Entrypoint")
  75         spkiHash := flag.String("spki", mmc.GetSPKIHash(), "Entrypoint's SPKI hash")
  76         notifyCmd := flag.String("notify", "cmd/notify", "Path to notification handler")
  77         heartbeatCh := flag.String("heartbeat-ch", "town-square", "Channel for heartbeating")
  78         flag.Parse()
  79         log.SetFlags(log.Lshortfile)
  80         log.SetOutput(os.Stdout)
  81         UmaskCur = syscall.Umask(0)
  82         syscall.Umask(UmaskCur)
  83
  84         mkFifo("debug")
  85         var err error
  86         DebugFd, err = os.OpenFile(
  87                 "debug", os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
  88         )
  89         if err != nil {
  90                 log.Fatalln(err)
  91         }
  92         defer DebugFd.Close()
  93
  94         entrypointURL, err := url.Parse(*entrypoint)
  95         if err != nil {
  96                 log.Fatalln(err)
  97         }
  98         login, password := netrc.Find(entrypointURL.Hostname())
  99         if login == "" || password == "" {
 100                 log.Fatalln("no credentials found for:", entrypointURL.Hostname())
 101         }
 102         c := model.NewAPIv4Client(*entrypoint)
 103         c.HTTPClient.Transport = &http.Transport{
 104                 Proxy:                 http.ProxyFromEnvironment,
 105                 ForceAttemptHTTP2:     true,
 106                 MaxIdleConns:          100,
 107                 IdleConnTimeout:       90 * time.Second,
 108                 TLSHandshakeTimeout:   10 * time.Second,
 109                 ExpectContinueTimeout: 1 * time.Second,
 110                 TLSClientConfig: &tls.Config{
 111                         ServerName:            entrypointURL.Hostname(),
 112                         InsecureSkipVerify:    true,
 113                         VerifyPeerCertificate: mmc.NewVerifyPeerCertificate(*spkiHash),
 114                 },
 115         }
 116
 117         c.Login(login, password)
 118         me, resp, err := c.GetMe("")
 119         if err != nil {
 120                 if DebugFd != nil {
 121                         spew.Fdump(DebugFd, resp)
 122                 }
 123                 log.Fatalln(err)
 124         }
 125         if DebugFd != nil {
 126                 spew.Fdump(DebugFd, me)
 127         }
 128         log.Println("logged in")
 129
 130         time.Sleep(mmc.SleepTime)
 131         teams, resp, err := c.GetTeamsForUser(me.Id, "")
 132         if err != nil {
 133                 if DebugFd != nil {
 134                         spew.Fdump(DebugFd, resp)
 135                 }
 136                 log.Fatalln(err)
 137         }
 138         if DebugFd != nil {
 139                 spew.Fdump(DebugFd, teams)
 140         }
 141         Team := teams[0]
 142
 143         var updateQueue []string
 144         Chans := make(map[string]*model.Channel)
 145         time.Sleep(mmc.SleepTime)
 146         page, resp, err := c.GetChannelsForTeamForUser(Team.Id, me.Id, false, "")
 147         if err != nil {
 148                 if DebugFd != nil {
 149                         spew.Fdump(DebugFd, resp)
 150                 }
 151                 log.Fatalln(err)
 152         }
 153         if DebugFd != nil {
 154                 spew.Fdump(DebugFd, page)
 155         }
 156         for _, ch := range page {
 157                 if ch.Type == "D" {
 158                         continue
 159                 }
 160                 Chans[ch.Name] = ch
 161                 pth := path.Join("chans", strings.ReplaceAll(ch.Name, ".", "_"))
 162                 updateQueue = append(updateQueue, pth, ch.Id)
 163                 os.MkdirAll(pth, 0777)
 164                 rewriteIfChanged(path.Join(pth, "id"), ch.Id+"\n")
 165                 rewriteIfChanged(path.Join(pth, "info"), fmt.Sprintf(
 166                         "%s\n%s\n%s\n",
 167                         ch.DisplayName,
 168                         ch.Header,
 169                         ch.Purpose,
 170                 ))
 171                 if _, err := os.Stat(path.Join(pth, mmc.OutRec)); err != nil &&
 172                         errors.Is(err, fs.ErrNotExist) {
 173                         if _, err = os.OpenFile(
 174                                 path.Join(pth, mmc.OutRec), os.O_WRONLY|os.O_CREATE, 0o666,
 175                         ); err != nil {
 176                                 log.Fatalln(err)
 177                         }
 178                 }
 179
 180                 usersPth := path.Join(pth, "users")
 181                 mkFifo(usersPth)
 182                 go func(ch *model.Channel) {
 183                         for {
 184                                 time.Sleep(mmc.SleepTime)
 185                                 fd, err := os.OpenFile(
 186                                         usersPth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
 187                                 )
 188                                 if err != nil {
 189                                         log.Println("OpenFile:", usersPth, err)
 190                                         continue
 191                                 }
 192                                 for n := 0; ; n++ {
 193                                         users, resp, err := c.GetUsersInChannel(ch.Id, n, mmc.PerPage, "")
 194                                         if err != nil {
 195                                                 if DebugFd != nil {
 196                                                         spew.Fdump(DebugFd, resp)
 197                                                 }
 198                                                 log.Println("GetUsersInChannel:", err)
 199                                                 fd.Close()
 200                                                 continue
 201                                         }
 202                                         if DebugFd != nil {
 203                                                 spew.Fdump(DebugFd, users)
 204                                         }
 205                                         for _, u := range users {
 206                                                 fmt.Fprintf(fd, "%s\n", u.Username)
 207                                         }
 208                                         if len(users) < mmc.PerPage {
 209                                                 break
 210                                         }
 211                                 }
 212                                 fd.Close()
 213                         }
 214                 }(ch)
 215
 216                 pth = path.Join(pth, "in")
 217                 mkFifo(pth)
 218                 go func(ch *model.Channel) {
 219                         for {
 220                                 fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
 221                                 if err != nil {
 222                                         continue
 223                                 }
 224                                 data, err := io.ReadAll(fd)
 225                                 fd.Close()
 226                                 if err != nil {
 227                                         continue
 228                                 }
 229                                 if _, err = makePost(c, ch.Id, string(data)); err != nil {
 230                                         log.Println("makePost:", err)
 231                                 }
 232                         }
 233                 }(ch)
 234         }
 235
 236         Users, err := mmc.GetUsers(c, DebugFd)
 237         if err != nil {
 238                 log.Fatalln(err)
 239         }
 240
 241         UsersDC := make(map[string]*model.Channel, len(Users))
 242         for _, u := range Users {
 243                 pth := path.Join("users", strings.ReplaceAll(u.Username, ".", "_"))
 244                 os.MkdirAll(pth, 0777)
 245                 rewriteIfChanged(
 246                         path.Join(pth, "name"),
 247                         fmt.Sprintf("%s %s\n", u.FirstName, u.LastName),
 248                 )
 249                 rewriteIfChanged(path.Join(pth, "email"), u.Email+"\n")
 250                 rewriteIfChanged(path.Join(pth, "id"), u.Id+"\n")
 251                 if _, err := os.Stat(path.Join(pth, mmc.OutRec)); err != nil &&
 252                         errors.Is(err, fs.ErrNotExist) {
 253                         if _, err = os.OpenFile(
 254                                 path.Join(pth, mmc.OutRec), os.O_WRONLY|os.O_CREATE, 0o666,
 255                         ); err != nil {
 256                                 log.Fatalln(err)
 257                         }
 258                 }
 259
 260                 if fi, err := os.Stat(path.Join(pth, mmc.OutRec)); err == nil && fi.Size() > 0 {
 261                         time.Sleep(mmc.SleepTime)
 262                         dc, resp, err := c.CreateDirectChannel(me.Id, u.Id)
 263                         if err != nil {
 264                                 if DebugFd != nil {
 265                                         spew.Fdump(DebugFd, resp)
 266                                 }
 267                                 log.Println("CreateDirectChannel:", err)
 268                                 continue
 269                         }
 270                         if DebugFd != nil {
 271                                 spew.Fdump(DebugFd, dc)
 272                         }
 273                         UsersDC[u.Id] = dc
 274                         updateQueue = append(updateQueue, pth, dc.Id)
 275                 }
 276
 277                 statusPth := path.Join(pth, "status")
 278                 mkFifo(statusPth)
 279                 go func(u *model.User) {
 280                         for {
 281                                 time.Sleep(mmc.SleepTime)
 282                                 fd, err := os.OpenFile(
 283                                         statusPth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
 284                                 )
 285                                 if err != nil {
 286                                         log.Println("OpenFile:", statusPth, err)
 287                                         continue
 288                                 }
 289                                 status, resp, err := c.GetUserStatus(u.Id, "")
 290                                 if err != nil {
 291                                         if DebugFd != nil {
 292                                                 spew.Fdump(DebugFd, resp)
 293                                         }
 294                                         log.Println("GetUserStatus:", err)
 295                                         fd.Close()
 296                                         continue
 297                                 }
 298                                 if DebugFd != nil {
 299                                         spew.Fdump(DebugFd, status)
 300                                 }
 301                                 fmt.Fprintf(fd, "%s\n", status.Status)
 302                                 fd.Close()
 303                         }
 304                 }(u)
 305
 306                 pth = path.Join(pth, "in")
 307                 mkFifo(pth)
 308                 go func(u *model.User) {
 309                         var dc *model.Channel
 310                         for {
 311                                 fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
 312                                 if err != nil {
 313                                         continue
 314                                 }
 315                                 data, err := io.ReadAll(fd)
 316                                 fd.Close()
 317                                 if err != nil {
 318                                         continue
 319                                 }
 320                                 dc = UsersDC[u.Id]
 321                                 if dc == nil {
 322                                         dc, resp, err = c.CreateDirectChannel(me.Id, u.Id)
 323                                         if err != nil {
 324                                                 if DebugFd != nil {
 325                                                         spew.Fdump(DebugFd, resp)
 326                                                 }
 327                                                 log.Println("CreateDirectChannel:", err)
 328                                                 continue
 329                                         }
 330                                         UsersDC[u.Id] = dc
 331                                         if DebugFd != nil {
 332                                                 spew.Fdump(DebugFd, dc)
 333                                         }
 334                                 }
 335                                 if _, err = makePost(c, dc.Id, string(data)); err != nil {
 336                                         log.Println("makePost:", err)
 337                                 }
 338                         }
 339                 }(u)
 340         }
 341
 342         UserStatus := make(map[string]string)
 343         var UserStatusM sync.RWMutex
 344         go func() {
 345                 pth := path.Join("users", "status")
 346                 mkFifo(pth)
 347                 for {
 348                         time.Sleep(mmc.SleepTime)
 349                         fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
 350                         if err != nil {
 351                                 log.Println("OpenFile:", pth, err)
 352                                 continue
 353                         }
 354                         agg := make(map[string][]string)
 355                         UserStatusM.RLock()
 356                         for name, status := range UserStatus {
 357                                 agg[status] = append(agg[status], name)
 358                         }
 359                         UserStatusM.RUnlock()
 360                         statuses := make([]string, 0, len(agg))
 361                         for status := range agg {
 362                                 sort.Strings(agg[status])
 363                                 statuses = append(statuses, status)
 364                         }
 365                         sort.Strings(statuses)
 366                         for _, status := range statuses {
 367                                 fmt.Fprintln(fd, status+":", strings.Join(agg[status], " "))
 368                         }
 369                         fd.Close()
 370                 }
 371         }()
 372
 373         log.Println("syncing", len(updateQueue)/2, "rooms")
 374         for len(updateQueue) > 0 {
 375                 err := updatePosts(c, Users, updateQueue[0], updateQueue[1])
 376                 if err != nil {
 377                         log.Println("updatePosts:", err)
 378                 }
 379                 updateQueue = updateQueue[2:]
 380         }
 381         log.Println("sync done")
 382         if *notifyCmd != "" {
 383                 exec.Command(*notifyCmd, "sync done").Run()
 384         }
 385
 386         go func() {
 387                 os.MkdirAll("file", 0777)
 388                 pthGet := path.Join("file", "get")
 389                 mkFifo(pthGet)
 390                 pthOut := path.Join("file", "out")
 391                 mkFifo(pthOut)
 392                 for {
 393                         time.Sleep(mmc.SleepTime)
 394                         fd, err := os.OpenFile(pthGet, os.O_RDONLY, os.FileMode(0666))
 395                         if err != nil {
 396                                 continue
 397                         }
 398                         data, err := io.ReadAll(fd)
 399                         fd.Close()
 400                         if err != nil {
 401                                 continue
 402                         }
 403                         fileId := strings.TrimRight(string(data), " \n")
 404                         if len(fileId) == 0 {
 405                                 continue
 406                         }
 407                         fi, resp, err := c.GetFileInfo(fileId)
 408                         if DebugFd != nil {
 409                                 spew.Fdump(DebugFd, fi, resp)
 410                         }
 411                         if err != nil {
 412                                 log.Println(err)
 413                                 continue
 414                         }
 415                         if fi == nil {
 416                                 fmt.Println(resp)
 417                                 continue
 418                         }
 419                         data, resp, err = c.GetFile(fileId)
 420                         if err != nil {
 421                                 if DebugFd != nil {
 422                                         spew.Fdump(DebugFd, resp)
 423                                 }
 424                                 log.Println(err)
 425                                 continue
 426                         }
 427                         fd, err = os.OpenFile(
 428                                 pthOut, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
 429                         )
 430                         if err != nil {
 431                                 log.Println(err)
 432                                 continue
 433                         }
 434                         w := tar.NewWriter(fd)
 435                         if err = w.WriteHeader(&tar.Header{
 436                                 Format:   tar.FormatPAX,
 437                                 Typeflag: tar.TypeReg,
 438                                 Name:     fi.Name,
 439                                 Size:     int64(len(data)),
 440                                 Mode:     0o666,
 441                                 ModTime:  time.Unix(fi.CreateAt/1000, 0),
 442                                 PAXRecords: map[string]string{
 443                                         "MM.FileId":   fi.Id,
 444                                         "MM.MIMEType": fi.MimeType,
 445                                 },
 446                         }); err != nil {
 447                                 log.Println(err)
 448                                 fd.Close()
 449                                 continue
 450                         }
 451                         if _, err = io.Copy(w, bytes.NewReader(data)); err != nil {
 452                                 log.Println(err)
 453                                 fd.Close()
 454                                 continue
 455                         }
 456                         err = w.Close()
 457                         fd.Close()
 458                         if err != nil {
 459                                 log.Println(err)
 460                         }
 461                 }
 462         }()
 463
 464         needsShutdown := make(chan os.Signal)
 465         switch entrypointURL.Scheme {
 466         case "http":
 467                 entrypointURL.Scheme = "ws"
 468         case "https":
 469                 entrypointURL.Scheme = "wss"
 470         default:
 471                 log.Println("unhandled scheme:", entrypointURL.Scheme)
 472         }
 473         wc, err := model.NewWebSocketClient4WithDialer(
 474                 &websocket.Dialer{
 475                         TLSClientConfig: &tls.Config{
 476                                 ServerName:            entrypointURL.Hostname(),
 477                                 InsecureSkipVerify:    true,
 478                                 VerifyPeerCertificate: mmc.NewVerifyPeerCertificate(*spkiHash),
 479                         },
 480                 }, entrypointURL.String(), c.AuthToken,
 481         )
 482         if err != nil {
 483                 log.Fatalln(err)
 484         }
 485         go func() {
 486                 wc.Listen()
 487                 wc.GetStatuses()
 488                 t := time.NewTicker(time.Minute)
 489                 for {
 490                         select {
 491                         case <-t.C:
 492                                 if wc.ListenError != nil {
 493                                         log.Println("ListenError:", wc.ListenError)
 494                                         needsShutdown <- syscall.SIGTERM
 495                                         return
 496                                 }
 497                                 wc.GetStatuses()
 498                                 if *heartbeatCh != "" {
 499                                         if _, _, err = c.ViewChannel(
 500                                                 me.Id,
 501                                                 &model.ChannelView{ChannelId: Chans[*heartbeatCh].Id},
 502                                         ); err != nil {
 503                                                 log.Println("ChannelView:", err)
 504                                         }
 505                                 }
 506                         case <-wc.PingTimeoutChannel:
 507                                 log.Println("PING timeout")
 508                                 needsShutdown <- syscall.SIGTERM
 509                                 return
 510                         case e := <-wc.EventChannel:
 511                                 if e == nil || !e.IsValid() {
 512                                         continue
 513                                 }
 514                                 if DebugFd != nil {
 515                                         spew.Fdump(DebugFd, e)
 516                                 }
 517                                 data := e.GetData()
 518                                 var user *model.User
 519                                 if userId, ok := data["user_id"]; ok && userId.(string) != "" {
 520                                         user = Users[userId.(string)]
 521                                         if user == nil {
 522                                                 log.Println("unknown user for event:", e)
 523                                                 continue
 524                                         }
 525                                 }
 526                                 switch eventType := e.EventType(); eventType {
 527                                 case model.WebsocketEventTyping:
 528                                         if *notifyCmd != "" {
 529                                                 exec.Command(*notifyCmd, "typing: "+user.Username).Run()
 530                                         }
 531                                 case model.WebsocketEventPostEdited,
 532                                         model.WebsocketEventPostDeleted,
 533                                         model.WebsocketEventPosted:
 534                                         chName, ok := data["channel_name"].(string)
 535                                         if !ok {
 536                                                 continue
 537                                         }
 538                                         var post model.Post
 539                                         if err = json.NewDecoder(
 540                                                 strings.NewReader(data["post"].(string)),
 541                                         ).Decode(&post); err != nil {
 542                                                 log.Fatalln(err)
 543                                         }
 544                                         var recipient string
 545                                         switch model.ChannelType(data["channel_type"].(string)) {
 546                                         case model.ChannelTypeDirect:
 547                                                 userId := strings.TrimPrefix(chName, me.Id+"__")
 548                                                 userId = strings.TrimSuffix(userId, "__"+me.Id)
 549                                                 user := Users[userId]
 550                                                 if user == nil {
 551                                                         log.Println("unknown user:", userId)
 552                                                         continue
 553                                                 }
 554                                                 recipient = path.Join("users", user.Username)
 555                                         case model.ChannelTypeOpen:
 556                                                 fallthrough
 557                                         case model.ChannelTypePrivate:
 558                                                 fallthrough
 559                                         case model.ChannelTypeGroup:
 560                                                 recipient = path.Join("chans", chName)
 561                                         }
 562                                         if err = writePosts(
 563                                                 recipient, Users, []mmc.Post{{P: &post, E: eventType}},
 564                                         ); err != nil {
 565                                                 log.Fatalln(err)
 566                                         }
 567                                 case model.WebsocketEventStatusChange:
 568                                         status := data["status"].(string)
 569                                         UserStatusM.Lock()
 570                                         UserStatus[user.Username] = status
 571                                         UserStatusM.Unlock()
 572                                         if *notifyCmd != "" {
 573                                                 exec.Command(*notifyCmd, fmt.Sprintf(
 574                                                         "status: %s -> %s", user.Username, status,
 575                                                 )).Run()
 576                                         }
 577                                 case model.WebsocketEventHello:
 578                                 case model.WebsocketEventChannelViewed:
 579                                 default:
 580                                         log.Println(e)
 581                                 }
 582                         case resp := <-wc.ResponseChannel:
 583                                 if resp == nil || !resp.IsValid() {
 584                                         continue
 585                                 }
 586                                 if DebugFd != nil {
 587                                         spew.Fdump(DebugFd, resp)
 588                                 }
 589                                 statuses := make(map[string]string)
 590                                 for userId, status := range resp.Data {
 591                                         status, ok := status.(string)
 592                                         if !ok {
 593                                                 continue
 594                                         }
 595                                         user := Users[userId]
 596                                         if user == nil {
 597                                                 continue
 598                                         }
 599                                         statuses[user.Username] = status
 600                                 }
 601                                 if len(statuses) > 0 {
 602                                         UserStatusM.Lock()
 603                                         for u := range UserStatus {
 604                                                 delete(UserStatus, u)
 605                                         }
 606                                         for u, status := range statuses {
 607                                                 UserStatus[u] = status
 608                                         }
 609                                         UserStatusM.Unlock()
 610                                 }
 611                         }
 612                 }
 613         }()
 614
 615         signal.Notify(needsShutdown, syscall.SIGTERM, syscall.SIGINT)
 616         <-needsShutdown
 617         c.Logout()
 618         log.Println("finished")
 619 }