Fix & document change_email.py (#3331)

Author: fisx

changelog.d/5-internal/pr-3321-scim-email-script

@@ -1 +1 @@
- Script to bulk-change/-repair user's scim and brig email address
+Script to bulk-change/-repair user's scim and brig email address (#3321, #3331)

docs/src/understand/single-sign-on/understand/main.md

@@ -407,6 +407,22 @@ export SCIM_TOKEN_ID=$(echo $SCIM_TOKEN_FULL | jq -r .info.id)
 
 The SCIM token is now contained in the `SCIM_TOKEN` environment variable.
 
+EXPERT HINT: If you are a site operator and have the ok from the team admin, but not their password, you can try something like this (WARNING!  TEST THIS ON NON-PROD ACCOUNTS FIRST!):
+
+```
+export id="$(uuid -v4)"
+export token_="$(dd if=/dev/random bs=32 count=1 | base64)"
+export created_at='2023-05-30 11:15:00+0000'
+export team=...
+export descr=...
+echo 'INSERT INTO spar.team_provisioning_by_token (token_, team, id, created_at, idp, descr) VALUES ('\'$token_\', $team, $id, \'$created_at\', NULL, \'$descr\'');'
+echo 'INSERT INTO spar.team_provisioning_by_team (token_, team, id, created_at, idp, descr) VALUES ('\'$token_\', $team, $id, \'$created_at\', NULL, \'$descr\'');'
+echo 'SELECT * FROM spar.team_provisioning_by_team WHERE '"team = $team AND id = $id;"
+echo 'SELECT * FROM spar.team_provisioning_by_token WHERE '"token_ = $token_;"
+echo 'DELETE FROM spar.team_provisioning_by_team WHERE '"team = $team AND id = $id;"
+echo 'DELETE FROM spar.team_provisioning_by_token WHERE '"token_ = $token_;"
+```
+
 You can look it up again with:
 
 ```{code-block} bash

hack/bin/change_emails.py

@@ -42,13 +42,16 @@
 from wire import api
 from wire.context import Context
 import csv
+import time # for `time.sleep`.  FUTUREWORK: you can probably tweak the
+            # sleeps we've injected a lot, or remove all of them,
+            # without doing any harm.  just test it on staging first!
 import argparse
 
 ### configure this section
 
 scim_token = "..."
 filename = './test.csv'
-ctx = Context(domain="localhost", version="4", service_map={'spar': 8081, 'brig': 8082})
+ctx = None
 
 ### brig, spar api
 
@@ -72,61 +75,96 @@ def confirm_new_email(ctx, user_id, key, code):
     url = ctx.mkurl("brig", f"/activate")
     return ctx.request('GET', url, headers=({'Z-User': user_id}), params=({'key': key, 'code': code}))
 
+def assert_resp(resp, status_want):
+    if resp.response.status_code == status_want:
+        return
+    else:
+        print(resp.text);
+        # FUTUREWORK: it's not necessary to fail here; we could also
+        # printf this and continue with the next row of the input;
+        # later collecting all the failing cases into a list to be
+        # handed back to the team admin for further analysis.
+        assert False
+
 ### idioms
 
 def confirm_email(user_id, email):
     r = get_activation_code(ctx, user_id, email)
-    assert r.response.status_code == 200
+    assert_resp(r, 200)
     r2 = confirm_new_email(ctx, user_id, r.json()['key'], r.json()['code'])
-    assert r2.response.status_code == 200
+    assert_resp(r2, 200)
     return r2
 
+# update from one email address to another.  this is called twice,
+# first for "to old address" to make the db state consistent, and then
+# for "to new address".  it's possible that this should be two
+# functions, but so far they have no difference to suggest that.
 def update(user_id, email):
     r = get_scim_user(ctx, user_id)
-    assert r.response.status_code == 200
+    assert_resp(r, 200)
     body = dict(r.json())
     if body['externalId'] != email:
         body['externalId'] = email
         r2 = put_scim_user(ctx, user_id, body)
-        assert r2.response.status_code == 200
+        body2 = r2.json()
+        # FUTUREWORK: there are some legitimate ways to fail in the
+        # following line, if the email address is already the old one.
+        # add code here to your liking in order to handle these more
+        # gracefully.  you may want to distinguish between "change to
+        # old address to make everything consistent" (where it's ok if
+        # you find that it's already consistent, and you don't need to
+        # do anything) and "change to new email address" (where you
+        # certainly want a response that says "yes, it worked").
+        assert_resp(r2, 200)
         return True
     else:
         report_state('old=new', user_id)
         return False
 
 def report_state(msg, user_id):
     r = get_scim_user(ctx, user_id)
-    assert r.response.status_code == 200
+    assert_resp(r, 200)
     r2 = get_brig_user(ctx, user_id)
-    assert r.response.status_code == 200
+    assert_resp(r2, 200)
     print(f"[{msg}] uid={user_id}; scim_email={r.json()['externalId']}; brig_email={r2.json()['email']}")
+    return r.json()['externalId'], r2.json()['email']
+
 
 ### process one item
 
 def update_back_and_forth(user_id, old_email, new_email):
     assert old_email != new_email
 
     # nothing has happened yet
-    report_state('before', user_id)
+    email_scim, email_brig = report_state('before', user_id)
 
     # change to old email address to unblock pending confirmation
+    # if brig already has the old email, setting to old email won't yield an activation code, so we don't.
     if update(user_id, old_email):
-        confirm_email(user_id, old_email)
+        if email_brig != old_email:
+            time.sleep(4)
+            confirm_email(user_id, old_email)
     report_state('between', user_id)
 
     # change back to new email address
-    assert update(user_id, new_email)
-    confirm_email(user_id, new_email)
+    if update(user_id, new_email):
+        confirm_email(user_id, new_email)
+    else:
+        # nothing to confirm; logging happens in update_to_new
+        pass
     report_state('after', user_id)
 
 ### process csv file
 
 def main():
+    global ctx
     with open(filename, newline='') as csvfile:
         rows = csv.reader(csvfile, delimiter=',')
         for row in rows:
+            ctx = Context(domain="localhost", version="4", service_map={'spar': 8081, 'brig': 8082})
             [old_email, user_id, new_email] = row
             update_back_and_forth(user_id, old_email, new_email)
+            time.sleep(5)
 
 if __name__ == '__main__':
    main()