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