sdk/python/arvados/commands/federation_migrate.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) The Arvados Authors. All rights reserved.
   3 #
   4 # SPDX-License-Identifier: Apache-2.0
   5
   6 #
   7 # Migration tool for merging user accounts belonging to the same user
   8 # but on separate clusters to use a single user account managed by a
   9 # specific cluster.
  10 #
  11 # If you're working on this, see
  12 # arvados/sdk/python/tests/fed-migrate/README for information about
  13 # the testing infrastructure.
  14
  15 import arvados
  16 import arvados.util
  17 import arvados.errors
  18 import csv
  19 import sys
  20 import argparse
  21 import hmac
  22 import urllib.parse
  23 import os
  24 import hashlib
  25 from arvados._version import __version__
  26
  27 EMAIL=0
  28 USERNAME=1
  29 UUID=2
  30 HOMECLUSTER=3
  31
  32 def connect_clusters(args):
  33     clusters = {}
  34     errors = []
  35     loginCluster = None
  36     if args.tokens:
  37         print("Reading %s" % args.tokens)
  38         with open(args.tokens, "rt") as f:
  39             for r in csv.reader(f):
  40                 if len(r) != 2:
  41                     continue
  42                 host = r[0]
  43                 token = r[1]
  44                 print("Contacting %s" % (host))
  45                 arv = arvados.api(host=host, token=token, cache=False)
  46                 clusters[arv._rootDesc["uuidPrefix"]] = arv
  47     else:
  48         arv = arvados.api(cache=False)
  49         rh = arv._rootDesc["remoteHosts"]
  50         tok = arv.api_client_authorizations().current().execute()
  51         token = "v2/%s/%s" % (tok["uuid"], tok["api_token"])
  52
  53         for k,v in rh.items():
  54             arv = arvados.api(host=v, token=token, cache=False, insecure=os.environ.get("ARVADOS_API_HOST_INSECURE"))
  55             clusters[k] = arv
  56
  57     for _, arv in clusters.items():
  58         config = arv.configs().get().execute()
  59         if config["Login"]["LoginCluster"] != "" and loginCluster is None:
  60             loginCluster = config["Login"]["LoginCluster"]
  61
  62     print("Checking that the federation is well connected")
  63     for arv in clusters.values():
  64         config = arv.configs().get().execute()
  65         if loginCluster and config["Login"]["LoginCluster"] != loginCluster and config["ClusterID"] != loginCluster:
  66             errors.append("Inconsistent login cluster configuration, expected '%s' on %s but was '%s'" % (loginCluster, config["ClusterID"], config["Login"]["LoginCluster"]))
  67             continue
  68
  69         if arv._rootDesc["revision"] < "20190926":
  70             errors.append("Arvados API server revision on cluster '%s' is too old, must be updated to at least Arvados 1.5 before running migration." % config["ClusterID"])
  71             continue
  72
  73         try:
  74             cur = arv.users().current().execute()
  75         except arvados.errors.ApiError as e:
  76             errors.append("checking token for %s   %s" % (arv._rootDesc["rootUrl"], e))
  77             continue
  78
  79         if not cur["is_admin"]:
  80             errors.append("User %s is not admin on %s" % (cur["uuid"], arv._rootDesc["uuidPrefix"]))
  81             continue
  82
  83         for r in clusters:
  84             if r != arv._rootDesc["uuidPrefix"] and r not in arv._rootDesc["remoteHosts"]:
  85                 errors.append("%s is missing from remoteHosts of %s" % (r, arv._rootDesc["uuidPrefix"]))
  86         for r in arv._rootDesc["remoteHosts"]:
  87             if r != "*" and r not in clusters:
  88                 print("WARNING: %s is federated with %s but %s is missing from the tokens file or the token is invalid" % (arv._rootDesc["uuidPrefix"], r, r))
  89
  90     return clusters, errors, loginCluster
  91
  92
  93 def fetch_users(clusters, loginCluster):
  94     rows = []
  95     by_email = {}
  96     by_username = {}
  97
  98     users = []
  99     for c, arv in clusters.items():
 100         print("Getting user list from %s" % c)
 101         ul = arvados.util.list_all(arv.users().list)
 102         for l in ul:
 103             if l["uuid"].startswith(c):
 104                 users.append(l)
 105
 106     # Users list is sorted by email
 107     # Go through users and collect users with same email
 108     # when we see a different email (or get to the end)
 109     # call add_accum_rows() to generate the report rows with
 110     # the "home cluster" set, and also fill in the by_email table.
 111
 112     users = sorted(users, key=lambda u: u["email"]+"::"+(u["username"] or "")+"::"+u["uuid"])
 113
 114     accum = []
 115     lastemail = None
 116
 117     def add_accum_rows():
 118         homeuuid = None
 119         for a in accum:
 120             uuids = set(a["uuid"] for a in accum)
 121             homeuuid = ((len(uuids) == 1) and uuids.pop()) or ""
 122         for a in accum:
 123             r = (a["email"], a["username"], a["uuid"], loginCluster or homeuuid[0:5])
 124             by_email.setdefault(a["email"], {})
 125             by_email[a["email"]][a["uuid"]] = r
 126             homeuuid_and_username = "%s::%s" % (r[HOMECLUSTER], a["username"])
 127             if homeuuid_and_username not in by_username:
 128                 by_username[homeuuid_and_username] = a["email"]
 129             elif by_username[homeuuid_and_username] != a["email"]:
 130                 print("ERROR: the username '%s' is listed for both '%s' and '%s' on cluster '%s'" % (r[USERNAME], r[EMAIL], by_username[homeuuid_and_username], r[HOMECLUSTER]))
 131                 exit(1)
 132             rows.append(r)
 133
 134     for u in users:
 135         if u["uuid"].endswith("-anonymouspublic") or u["uuid"].endswith("-000000000000000"):
 136             continue
 137         if lastemail == None:
 138             lastemail = u["email"]
 139         if u["email"] == lastemail:
 140             accum.append(u)
 141         else:
 142             add_accum_rows()
 143             lastemail = u["email"]
 144             accum = [u]
 145
 146     add_accum_rows()
 147
 148     return rows, by_email, by_username
 149
 150
 151 def read_migrations(args, by_email, by_username):
 152     rows = []
 153     with open(args.migrate or args.dry_run, "rt") as f:
 154         for r in csv.reader(f):
 155             if r[EMAIL] == "email":
 156                 continue
 157             by_email.setdefault(r[EMAIL], {})
 158             by_email[r[EMAIL]][r[UUID]] = r
 159
 160             homeuuid_and_username = "%s::%s" % (r[HOMECLUSTER], r[USERNAME])
 161             if homeuuid_and_username not in by_username:
 162                 by_username[homeuuid_and_username] = r[EMAIL]
 163             elif by_username[homeuuid_and_username] != r[EMAIL]:
 164                 print("ERROR: the username '%s' is listed for both '%s' and '%s' on cluster '%s'" % (r[USERNAME], r[EMAIL], by_username[homeuuid_and_username], r[HOMECLUSTER]))
 165                 exit(1)
 166
 167             rows.append(r)
 168     return rows
 169
 170 def update_username(args, email, user_uuid, username, migratecluster, migratearv):
 171     print("(%s) Updating username of %s to '%s' on %s" % (email, user_uuid, username, migratecluster))
 172     if not args.dry_run:
 173         try:
 174             conflicts = migratearv.users().list(filters=[["username", "=", username]]).execute()
 175             if conflicts["items"]:
 176                 migratearv.users().update(uuid=conflicts["items"][0]["uuid"], body={"user": {"username": username+"migrate"}}).execute()
 177             migratearv.users().update(uuid=user_uuid, body={"user": {"username": username}}).execute()
 178         except arvados.errors.ApiError as e:
 179             print("(%s) Error updating username of %s to '%s' on %s: %s" % (email, user_uuid, username, migratecluster, e))
 180
 181
 182 def choose_new_user(args, by_email, email, userhome, username, old_user_uuid, clusters):
 183     candidates = []
 184     conflict = False
 185     for b in by_email[email].values():
 186         if b[2].startswith(userhome):
 187             candidates.append(b)
 188         if b[1] != username and b[3] == userhome:
 189             print("(%s) Cannot migrate %s, conflicting usernames %s and %s" % (email, old_user_uuid, b[1], username))
 190             conflict = True
 191             break
 192     if conflict:
 193         return None
 194     if len(candidates) == 0:
 195         if len(userhome) == 5 and userhome not in clusters:
 196             print("(%s) Cannot migrate %s, unknown home cluster %s (typo?)" % (email, old_user_uuid, userhome))
 197             return None
 198         print("(%s) No user listed with same email to migrate %s to %s, will create new user with username '%s'" % (email, old_user_uuid, userhome, username))
 199         if not args.dry_run:
 200             oldhomecluster = old_user_uuid[0:5]
 201             oldhomearv = clusters[oldhomecluster]
 202             newhomecluster = userhome[0:5]
 203             homearv = clusters[userhome]
 204             user = None
 205             try:
 206                 olduser = oldhomearv.users().get(uuid=old_user_uuid).execute()
 207                 conflicts = homearv.users().list(filters=[["username", "=", username]]).execute()
 208                 if conflicts["items"]:
 209                     homearv.users().update(uuid=conflicts["items"][0]["uuid"], body={"user": {"username": username+"migrate"}}).execute()
 210                 user = homearv.users().create(body={"user": {"email": email, "username": username, "is_active": olduser["is_active"]}}).execute()
 211             except arvados.errors.ApiError as e:
 212                 print("(%s) Could not create user: %s" % (email, str(e)))
 213                 return None
 214
 215             tup = (email, username, user["uuid"], userhome)
 216         else:
 217             # dry run
 218             tup = (email, username, "%s-tpzed-xfakexfakexfake" % (userhome[0:5]), userhome)
 219         by_email[email][tup[2]] = tup
 220         candidates.append(tup)
 221     if len(candidates) > 1:
 222         print("(%s) Multiple users listed to migrate %s to %s, use full uuid" % (email, old_user_uuid, userhome))
 223         return None
 224     return candidates[0][2]
 225
 226
 227 def activate_remote_user(args, email, homearv, migratearv, old_user_uuid, new_user_uuid):
 228     # create a token for the new user and salt it for the
 229     # migration cluster, then use it to access the migration
 230     # cluster as the new user once before merging to ensure
 231     # the new user is known on that cluster.
 232     migratecluster = migratearv._rootDesc["uuidPrefix"]
 233     try:
 234         if not args.dry_run:
 235             newtok = homearv.api_client_authorizations().create(body={
 236                 "api_client_authorization": {'owner_uuid': new_user_uuid}}).execute()
 237         else:
 238             newtok = {"uuid": "dry-run", "api_token": "12345"}
 239     except arvados.errors.ApiError as e:
 240         print("(%s) Could not create API token for %s: %s" % (email, new_user_uuid, e))
 241         return None
 242
 243     try:
 244         olduser = migratearv.users().get(uuid=old_user_uuid).execute()
 245     except arvados.errors.ApiError as e:
 246         if e.resp.status != 404:
 247             print("(%s) Could not retrieve user %s from %s, user may have already been migrated: %s" % (email, old_user_uuid, migratecluster, e))
 248         return None
 249
 250     salted = 'v2/' + newtok["uuid"] + '/' + hmac.new(newtok["api_token"].encode(),
 251                                                      msg=migratecluster.encode(),
 252                                                      digestmod=hashlib.sha1).hexdigest()
 253     try:
 254         ru = urllib.parse.urlparse(migratearv._rootDesc["rootUrl"])
 255         if not args.dry_run:
 256             newuser = arvados.api(host=ru.netloc, token=salted, insecure=os.environ.get("ARVADOS_API_HOST_INSECURE")).users().current().execute()
 257         else:
 258             newuser = {"is_active": True, "username": username}
 259     except arvados.errors.ApiError as e:
 260         print("(%s) Error getting user info for %s from %s: %s" % (email, new_user_uuid, migratecluster, e))
 261         return None
 262
 263     if not newuser["is_active"] and olduser["is_active"]:
 264         print("(%s) Activating user %s on %s" % (email, new_user_uuid, migratecluster))
 265         try:
 266             if not args.dry_run:
 267                 migratearv.users().update(uuid=new_user_uuid, body={"is_active": True}).execute()
 268         except arvados.errors.ApiError as e:
 269             print("(%s) Could not activate user %s on %s: %s" % (email, new_user_uuid, migratecluster, e))
 270             return None
 271
 272     if olduser["is_admin"] and not newuser["is_admin"]:
 273         print("(%s) Not migrating %s because user is admin but target user %s is not admin on %s" % (email, old_user_uuid, new_user_uuid, migratecluster))
 274         return None
 275
 276     return newuser
 277
 278 def migrate_user(args, migratearv, email, new_user_uuid, old_user_uuid):
 279     try:
 280         if not args.dry_run:
 281             grp = migratearv.groups().create(body={
 282                 "owner_uuid": new_user_uuid,
 283                 "name": "Migrated from %s (%s)" % (email, old_user_uuid),
 284                 "group_class": "project"
 285             }, ensure_unique_name=True).execute()
 286             migratearv.users().merge(old_user_uuid=old_user_uuid,
 287                                      new_user_uuid=new_user_uuid,
 288                                      new_owner_uuid=grp["uuid"],
 289                                      redirect_to_new_user=True).execute()
 290     except arvados.errors.ApiError as e:
 291         print("(%s) Error migrating user: %s" % (email, e))
 292
 293
 294 def main():
 295
 296     parser = argparse.ArgumentParser(description='Migrate users to federated identity, see https://doc.arvados.org/admin/merge-remote-account.html')
 297     parser.add_argument(
 298         '--version', action='version', version="%s %s" % (sys.argv[0], __version__),
 299         help='Print version and exit.')
 300     parser.add_argument('--tokens', type=str, required=False)
 301     group = parser.add_mutually_exclusive_group(required=True)
 302     group.add_argument('--report', type=str, help="Generate report .csv file listing users by email address and their associated Arvados accounts")
 303     group.add_argument('--migrate', type=str, help="Consume report .csv and migrate users to designated Arvados accounts")
 304     group.add_argument('--dry-run', type=str, help="Consume report .csv and report how user would be migrated to designated Arvados accounts")
 305     group.add_argument('--check', action="store_true", help="Check that tokens are usable and the federation is well connected")
 306     args = parser.parse_args()
 307
 308     clusters, errors, loginCluster = connect_clusters(args)
 309
 310     if errors:
 311         for e in errors:
 312             print("ERROR: "+str(e))
 313         exit(1)
 314
 315     if args.check:
 316         print("Tokens file passed checks")
 317         exit(0)
 318
 319     rows, by_email, by_username = fetch_users(clusters, loginCluster)
 320
 321     if args.report:
 322         out = csv.writer(open(args.report, "wt"))
 323         out.writerow(("email", "username", "user uuid", "home cluster"))
 324         for r in rows:
 325             out.writerow(r)
 326         print("Wrote %s" % args.report)
 327         return
 328
 329     if args.migrate or args.dry_run:
 330         if args.dry_run:
 331             print("Performing dry run")
 332
 333         rows = read_migrations(args, by_email, by_username)
 334
 335         for r in rows:
 336             email = r[EMAIL]
 337             username = r[USERNAME]
 338             old_user_uuid = r[UUID]
 339             userhome = r[HOMECLUSTER]
 340
 341             if userhome == "":
 342                 print("(%s) Skipping %s, no home cluster specified" % (email, old_user_uuid))
 343             if old_user_uuid.startswith(userhome):
 344                 migratecluster = old_user_uuid[0:5]
 345                 migratearv = clusters[migratecluster]
 346                 if migratearv.users().get(uuid=old_user_uuid).execute()["username"] != username:
 347                     update_username(args, email, old_user_uuid, username, migratecluster, migratearv)
 348                 continue
 349
 350             new_user_uuid = choose_new_user(args, by_email, email, userhome, username, old_user_uuid, clusters)
 351             if new_user_uuid is None:
 352                 continue
 353
 354             # cluster where the migration is happening
 355             for migratecluster in clusters:
 356                 migratearv = clusters[migratecluster]
 357
 358                 # the user's new home cluster
 359                 newhomecluster = userhome[0:5]
 360                 homearv = clusters[newhomecluster]
 361
 362                 newuser = activate_remote_user(args, email, homearv, migratearv, old_user_uuid, new_user_uuid)
 363                 if newuser is None:
 364                     continue
 365
 366                 print("(%s) Migrating %s to %s on %s" % (email, old_user_uuid, new_user_uuid, migratecluster))
 367
 368                 migrate_user(args, migratearv, email, new_user_uuid, old_user_uuid)
 369
 370                 if newuser['username'] != username:
 371                     update_username(args, email, new_user_uuid, username, migratecluster, migratearv)
 372
 373 if __name__ == "__main__":
 374     main()