tools/sync-users/sync-users.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "encoding/csv"
  10         "encoding/json"
  11         "flag"
  12         "fmt"
  13         "io"
  14         "log"
  15         "net/url"
  16         "os"
  17         "regexp"
  18         "strconv"
  19         "strings"
  20
  21         "git.arvados.org/arvados.git/lib/cmd"
  22         "git.arvados.org/arvados.git/sdk/go/arvados"
  23 )
  24
  25 var version = "dev"
  26
  27 type resourceList interface {
  28         Len() int
  29         GetItems() []interface{}
  30 }
  31
  32 // UserList implements resourceList interface
  33 type UserList struct {
  34         arvados.UserList
  35 }
  36
  37 // Len returns the amount of items this list holds
  38 func (l UserList) Len() int {
  39         return len(l.Items)
  40 }
  41
  42 // GetItems returns the list of items
  43 func (l UserList) GetItems() (out []interface{}) {
  44         for _, item := range l.Items {
  45                 out = append(out, item)
  46         }
  47         return
  48 }
  49
  50 func main() {
  51         cfg, err := GetConfig()
  52         if err != nil {
  53                 log.Fatalf("%v", err)
  54         }
  55
  56         if err := doMain(&cfg); err != nil {
  57                 log.Fatalf("%v", err)
  58         }
  59 }
  60
  61 type ConfigParams struct {
  62         Client             *arvados.Client
  63         ClusterID          string
  64         CurrentUser        arvados.User
  65         DeactivateUnlisted bool
  66         Path               string
  67         SysUserUUID        string
  68         AnonUserUUID       string
  69         Verbose            bool
  70 }
  71
  72 func ParseFlags(cfg *ConfigParams) error {
  73         flags := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
  74         flags.Usage = func() {
  75                 usageStr := `Synchronize remote users into Arvados from a CSV format file with 5 columns:
  76   * 1st: E-mail address
  77   * 2nd: First name
  78   * 3rd: Last name
  79   * 4th: Active status (0 or 1)
  80   * 5th: Admin status (0 or 1)`
  81                 fmt.Fprintf(flags.Output(), "%s\n\n", usageStr)
  82                 fmt.Fprintf(flags.Output(), "Usage:\n%s [OPTIONS] <input-file.csv>\n\n", os.Args[0])
  83                 fmt.Fprintf(flags.Output(), "Options:\n")
  84                 flags.PrintDefaults()
  85         }
  86
  87         deactivateUnlisted := flags.Bool(
  88                 "deactivate-unlisted",
  89                 false,
  90                 "Deactivate users that are not in the input file.")
  91         verbose := flags.Bool(
  92                 "verbose",
  93                 false,
  94                 "Log informational messages. Off by default.")
  95         getVersion := flags.Bool(
  96                 "version",
  97                 false,
  98                 "Print version information and exit.")
  99
 100         if ok, code := cmd.ParseFlags(flags, os.Args[0], os.Args[1:], "input-file.csv", os.Stderr); !ok {
 101                 os.Exit(code)
 102         } else if *getVersion {
 103                 fmt.Printf("%s %s\n", os.Args[0], version)
 104                 os.Exit(0)
 105         }
 106
 107         // Input file as a required positional argument
 108         if flags.NArg() == 0 {
 109                 return fmt.Errorf("please provide a path to an input file")
 110         }
 111         srcPath := &os.Args[flags.NFlag()+1]
 112
 113         // Validations
 114         if *srcPath == "" {
 115                 return fmt.Errorf("input file path invalid")
 116         }
 117
 118         cfg.DeactivateUnlisted = *deactivateUnlisted
 119         cfg.Path = *srcPath
 120         cfg.Verbose = *verbose
 121
 122         return nil
 123 }
 124
 125 // GetConfig sets up a ConfigParams struct
 126 func GetConfig() (cfg ConfigParams, err error) {
 127         err = ParseFlags(&cfg)
 128         if err != nil {
 129                 return
 130         }
 131
 132         cfg.Client = arvados.NewClientFromEnv()
 133
 134         // Check current user permissions
 135         u, err := cfg.Client.CurrentUser()
 136         if err != nil {
 137                 return cfg, fmt.Errorf("error getting the current user: %s", err)
 138         }
 139         if !u.IsAdmin {
 140                 return cfg, fmt.Errorf("current user %q is not an admin user", u.UUID)
 141         }
 142         if cfg.Verbose {
 143                 log.Printf("Running as admin user %q (%s)", u.Email, u.UUID)
 144         }
 145         cfg.CurrentUser = u
 146
 147         var ac struct {
 148                 ClusterID string
 149                 Login     struct {
 150                         LoginCluster string
 151                 }
 152         }
 153         err = cfg.Client.RequestAndDecode(&ac, "GET", "arvados/v1/config", nil, nil)
 154         if err != nil {
 155                 return cfg, fmt.Errorf("error getting the exported config: %s", err)
 156         }
 157         if ac.Login.LoginCluster != "" && ac.Login.LoginCluster != ac.ClusterID {
 158                 return cfg, fmt.Errorf("cannot run on a cluster other than the login cluster")
 159         }
 160         cfg.SysUserUUID = ac.ClusterID + "-tpzed-000000000000000"
 161         cfg.AnonUserUUID = ac.ClusterID + "-tpzed-anonymouspublic"
 162         cfg.ClusterID = ac.ClusterID
 163
 164         return cfg, nil
 165 }
 166
 167 func doMain(cfg *ConfigParams) error {
 168         // Try opening the input file early, just in case there's a problem.
 169         f, err := os.Open(cfg.Path)
 170         if err != nil {
 171                 return fmt.Errorf("error opening input file: %s", err)
 172         }
 173         defer f.Close()
 174
 175         allUsers := make(map[string]arvados.User)
 176         dupedEmails := make(map[string][]arvados.User)
 177         processedUsers := make(map[string]bool)
 178         results, err := GetAll(cfg.Client, "users", arvados.ResourceListParams{}, &UserList{})
 179         if err != nil {
 180                 return fmt.Errorf("error getting all users: %s", err)
 181         }
 182         log.Printf("Found %d users in cluster %q", len(results), cfg.ClusterID)
 183         localUserUuidRegex := regexp.MustCompile(fmt.Sprintf("^%s-tpzed-[0-9a-z]{15}$", cfg.ClusterID))
 184         for _, item := range results {
 185                 u := item.(arvados.User)
 186                 // Remote user check
 187                 if !localUserUuidRegex.MatchString(u.UUID) {
 188                         if cfg.Verbose {
 189                                 log.Printf("Remote user %q (%s) won't be considered for processing", u.Email, u.UUID)
 190                         }
 191                         continue
 192                 }
 193                 // Duplicated user's email check
 194                 email := strings.ToLower(u.Email)
 195                 if ul, ok := dupedEmails[email]; ok {
 196                         log.Printf("Duplicated email %q found in user %s", email, u.UUID)
 197                         dupedEmails[email] = append(ul, u)
 198                         continue
 199                 }
 200                 if eu, ok := allUsers[email]; ok {
 201                         log.Printf("Duplicated email %q found in users %s and %s", email, eu.UUID, u.UUID)
 202                         dupedEmails[email] = []arvados.User{eu, u}
 203                         delete(allUsers, email)
 204                         continue
 205                 }
 206                 allUsers[email] = u
 207                 processedUsers[email] = false
 208         }
 209
 210         loadedRecords, err := LoadInputFile(f)
 211         if err != nil {
 212                 return fmt.Errorf("reading input file %q: %s", cfg.Path, err)
 213         }
 214         log.Printf("Loaded %d records from input file", len(loadedRecords))
 215
 216         updatesSucceeded := map[string]bool{}
 217         updatesFailed := map[string]bool{}
 218         updatesSkipped := map[string]bool{}
 219
 220         for _, record := range loadedRecords {
 221                 processedUsers[record.Email] = true
 222                 if record.Email == cfg.CurrentUser.Email {
 223                         updatesSkipped[record.Email] = true
 224                         log.Printf("Skipping current user %q (%s) from processing", record.Email, cfg.CurrentUser.UUID)
 225                         continue
 226                 }
 227                 if updated, err := ProcessRecord(cfg, record, allUsers); err != nil {
 228                         log.Printf("error processing record %q: %s", record.Email, err)
 229                         updatesFailed[record.Email] = true
 230                 } else if updated {
 231                         updatesSucceeded[record.Email] = true
 232                 }
 233         }
 234
 235         if cfg.DeactivateUnlisted {
 236                 for email, user := range allUsers {
 237                         if shouldSkip(cfg, user) {
 238                                 updatesSkipped[email] = true
 239                                 log.Printf("Skipping unlisted user %q (%s) from deactivating", user.Email, user.UUID)
 240                                 continue
 241                         }
 242                         if !processedUsers[email] && allUsers[email].IsActive {
 243                                 if cfg.Verbose {
 244                                         log.Printf("Deactivating unlisted user %q (%s)", user.Email, user.UUID)
 245                                 }
 246                                 var updatedUser arvados.User
 247                                 if err := UnsetupUser(cfg.Client, user.UUID, &updatedUser); err != nil {
 248                                         log.Printf("error deactivating unlisted user %q: %s", user.UUID, err)
 249                                         updatesFailed[email] = true
 250                                 } else {
 251                                         allUsers[email] = updatedUser
 252                                         updatesSucceeded[email] = true
 253                                 }
 254                         }
 255                 }
 256         }
 257
 258         log.Printf("User update successes: %d, skips: %d, failures: %d", len(updatesSucceeded), len(updatesSkipped), len(updatesFailed))
 259
 260         // Report duplicated emails detection
 261         if len(dupedEmails) > 0 {
 262                 emails := make([]string, len(dupedEmails))
 263                 i := 0
 264                 for e := range dupedEmails {
 265                         emails[i] = e
 266                         i++
 267                 }
 268                 return fmt.Errorf("skipped %d duplicated email address(es) in the cluster's local user list: %v", len(dupedEmails), emails)
 269         }
 270
 271         return nil
 272 }
 273
 274 func shouldSkip(cfg *ConfigParams, user arvados.User) bool {
 275         switch user.UUID {
 276         case cfg.SysUserUUID, cfg.AnonUserUUID:
 277                 return true
 278         case cfg.CurrentUser.UUID:
 279                 return true
 280         }
 281         return false
 282 }
 283
 284 type userRecord struct {
 285         Email     string
 286         FirstName string
 287         LastName  string
 288         Active    bool
 289         Admin     bool
 290 }
 291
 292 // ProcessRecord creates or updates a user based on the given record
 293 func ProcessRecord(cfg *ConfigParams, record userRecord, allUsers map[string]arvados.User) (bool, error) {
 294         if cfg.Verbose {
 295                 log.Printf("Processing record for user %q", record.Email)
 296         }
 297
 298         wantedActiveStatus := strconv.FormatBool(record.Active)
 299         wantedAdminStatus := strconv.FormatBool(record.Admin)
 300         createRequired := false
 301         updateRequired := false
 302         // Check if user exists, set its active & admin status.
 303         var user arvados.User
 304         user, ok := allUsers[record.Email]
 305         if !ok {
 306                 if cfg.Verbose {
 307                         log.Printf("User %q does not exist, creating", record.Email)
 308                 }
 309                 createRequired = true
 310                 err := CreateUser(cfg.Client, &user, map[string]string{
 311                         "email":      record.Email,
 312                         "first_name": record.FirstName,
 313                         "last_name":  record.LastName,
 314                         "is_active":  wantedActiveStatus,
 315                         "is_admin":   wantedAdminStatus,
 316                 })
 317                 if err != nil {
 318                         return false, fmt.Errorf("error creating user %q: %s", record.Email, err)
 319                 }
 320         }
 321         if record.Active != user.IsActive {
 322                 updateRequired = true
 323                 if record.Active {
 324                         if cfg.Verbose {
 325                                 log.Printf("User %q is inactive, activating", record.Email)
 326                         }
 327                         // Here we assume the 'setup' is done elsewhere if needed.
 328                         err := UpdateUser(cfg.Client, user.UUID, &user, map[string]string{
 329                                 "is_active": wantedActiveStatus,
 330                                 "is_admin":  wantedAdminStatus, // Just in case it needs to be changed.
 331                         })
 332                         if err != nil {
 333                                 return false, fmt.Errorf("error updating user %q: %s", record.Email, err)
 334                         }
 335                 } else {
 336                         if cfg.Verbose {
 337                                 log.Printf("User %q is active, deactivating", record.Email)
 338                         }
 339                         err := UnsetupUser(cfg.Client, user.UUID, &user)
 340                         if err != nil {
 341                                 return false, fmt.Errorf("error deactivating user %q: %s", record.Email, err)
 342                         }
 343                 }
 344         }
 345         // Inactive users cannot be admins.
 346         if user.IsActive && record.Admin != user.IsAdmin {
 347                 if cfg.Verbose {
 348                         log.Printf("User %q is active, changing admin status to %v", record.Email, record.Admin)
 349                 }
 350                 updateRequired = true
 351                 err := UpdateUser(cfg.Client, user.UUID, &user, map[string]string{
 352                         "is_admin": wantedAdminStatus,
 353                 })
 354                 if err != nil {
 355                         return false, fmt.Errorf("error updating user %q: %s", record.Email, err)
 356                 }
 357         }
 358         allUsers[record.Email] = user
 359         if createRequired {
 360                 log.Printf("Created user %q", record.Email)
 361         }
 362         if updateRequired {
 363                 log.Printf("Updated user %q", record.Email)
 364         }
 365
 366         return createRequired || updateRequired, nil
 367 }
 368
 369 // LoadInputFile reads the input file and returns a list of user records
 370 func LoadInputFile(f *os.File) (loadedRecords []userRecord, err error) {
 371         lineNo := 0
 372         csvReader := csv.NewReader(f)
 373         loadedRecords = make([]userRecord, 0)
 374
 375         for {
 376                 record, e := csvReader.Read()
 377                 if e == io.EOF {
 378                         break
 379                 }
 380                 lineNo++
 381                 if e != nil {
 382                         err = fmt.Errorf("parsing error at line %d: %s", lineNo, e)
 383                         return
 384                 }
 385                 if len(record) != 5 {
 386                         err = fmt.Errorf("parsing error at line %d: expected 5 fields, found %d", lineNo, len(record))
 387                         return
 388                 }
 389                 email := strings.ToLower(strings.TrimSpace(record[0]))
 390                 firstName := strings.TrimSpace(record[1])
 391                 lastName := strings.TrimSpace(record[2])
 392                 active := strings.TrimSpace(record[3])
 393                 admin := strings.TrimSpace(record[4])
 394                 if email == "" || firstName == "" || lastName == "" || active == "" || admin == "" {
 395                         err = fmt.Errorf("parsing error at line %d: fields cannot be empty", lineNo)
 396                         return
 397                 }
 398                 activeBool, err := strconv.ParseBool(active)
 399                 if err != nil {
 400                         return nil, fmt.Errorf("parsing error at line %d: active status not recognized", lineNo)
 401                 }
 402                 adminBool, err := strconv.ParseBool(admin)
 403                 if err != nil {
 404                         return nil, fmt.Errorf("parsing error at line %d: admin status not recognized", lineNo)
 405                 }
 406                 loadedRecords = append(loadedRecords, userRecord{
 407                         Email:     email,
 408                         FirstName: firstName,
 409                         LastName:  lastName,
 410                         Active:    activeBool,
 411                         Admin:     adminBool,
 412                 })
 413         }
 414         return loadedRecords, nil
 415 }
 416
 417 // GetAll adds all objects of type 'resource' to the 'allItems' list
 418 func GetAll(c *arvados.Client, res string, params arvados.ResourceListParams, page resourceList) (allItems []interface{}, err error) {
 419         // Use the maximum page size the server allows
 420         limit := 1<<31 - 1
 421         params.Limit = &limit
 422         params.Offset = 0
 423         params.Order = "uuid"
 424         for {
 425                 if err = GetResourceList(c, &page, res, params); err != nil {
 426                         return allItems, err
 427                 }
 428                 // Have we finished paging?
 429                 if page.Len() == 0 {
 430                         break
 431                 }
 432                 allItems = append(allItems, page.GetItems()...)
 433                 params.Offset += page.Len()
 434         }
 435         return allItems, nil
 436 }
 437
 438 func jsonReader(rscName string, ob interface{}) io.Reader {
 439         j, err := json.Marshal(ob)
 440         if err != nil {
 441                 panic(err)
 442         }
 443         v := url.Values{}
 444         v[rscName] = []string{string(j)}
 445         return bytes.NewBufferString(v.Encode())
 446 }
 447
 448 // GetResourceList fetches res list using params
 449 func GetResourceList(c *arvados.Client, dst *resourceList, res string, params interface{}) error {
 450         return c.RequestAndDecode(dst, "GET", "/arvados/v1/"+res, nil, params)
 451 }
 452
 453 // CreateUser creates a user with userData parameters, assigns it to dst
 454 func CreateUser(c *arvados.Client, dst *arvados.User, userData map[string]string) error {
 455         return c.RequestAndDecode(dst, "POST", "/arvados/v1/users", jsonReader("user", userData), nil)
 456 }
 457
 458 // UpdateUser updates a user with userData parameters
 459 func UpdateUser(c *arvados.Client, userUUID string, dst *arvados.User, userData map[string]string) error {
 460         return c.RequestAndDecode(&dst, "PUT", "/arvados/v1/users/"+userUUID, jsonReader("user", userData), nil)
 461 }
 462
 463 // UnsetupUser deactivates a user
 464 func UnsetupUser(c *arvados.Client, userUUID string, dst *arvados.User) error {
 465         return c.RequestAndDecode(&dst, "POST", "/arvados/v1/users/"+userUUID+"/unsetup", nil, nil)
 466 }