From 0e0e3f72844c00f746685ad146e4c61645ee6c21 Mon Sep 17 00:00:00 2001 From: Brian Gregg Date: Wed, 29 Jul 2026 10:28:22 -0400 Subject: [PATCH] Added duplicate field check and Google Sheet protections check. Created a separate script to check a Google Sheet for protections that hinder updating by the service account. --- check_sheet_protections.py | 90 +++++++++++++++++++++++++ scan-batch-dir | 130 ++++++++++++++++++++++++++++++++++++- 2 files changed, 219 insertions(+), 1 deletion(-) create mode 100755 check_sheet_protections.py diff --git a/check_sheet_protections.py b/check_sheet_protections.py new file mode 100755 index 0000000..a28e4cf --- /dev/null +++ b/check_sheet_protections.py @@ -0,0 +1,90 @@ +#!/usr/bin/python3.12 +""" +Standalone diagnostic: list protected ranges/sheets on a Google Sheet, +and check whether a given service account email is allowed to bypass them. + +Usage: + python3 check_sheet_protections.py \ + --spreadsheet-id 1_nnBcCt4MPaAzaXqvAVD2pMv8AX9wPj8BBfgI-05Ri8 \ + --creds /path/to/configuration.json +""" + +import argparse +import json +from google.oauth2 import service_account +from googleapiclient.discovery import build + + +def connect(credentials_file): + scopes = ['https://www.googleapis.com/auth/spreadsheets.readonly'] + creds = service_account.Credentials.from_service_account_file( + credentials_file, scopes=scopes + ) + return build('sheets', 'v4', credentials=creds) + + +def get_service_account_email(credentials_file): + with open(credentials_file) as f: + data = json.load(f) + return data.get('client_email', 'UNKNOWN') + + +def describe_range(rng): + if not rng: + return "ENTIRE SHEET (no bounds set)" + parts = [] + if 'startRowIndex' in rng or 'endRowIndex' in rng: + parts.append(f"rows {rng.get('startRowIndex', '0')}-{rng.get('endRowIndex', 'end')}") + if 'startColumnIndex' in rng or 'endColumnIndex' in rng: + parts.append(f"cols {rng.get('startColumnIndex', '0')}-{rng.get('endColumnIndex', 'end')}") + return ", ".join(parts) if parts else "ENTIRE SHEET (no bounds set)" + + +def main(): + parser = argparse.ArgumentParser(description="Check Google Sheet protected ranges.") + parser.add_argument('--spreadsheet-id', required=True) + parser.add_argument('--creds', required=True, help="Path to service account credentials JSON.") + args = parser.parse_args() + + sa_email = get_service_account_email(args.creds) + print(f"Service account: {sa_email}\n") + + service = connect(args.creds) + result = service.spreadsheets().get( + spreadsheetId=args.spreadsheet_id, + fields="sheets(properties(sheetId,title),protectedRanges)" + ).execute() + + found_any = False + for sheet in result.get('sheets', []): + title = sheet['properties']['title'] + for pr in sheet.get('protectedRanges', []): + found_any = True + editors = pr.get('editors', {}) + editor_users = editors.get('users', []) + domain_can_edit = editors.get('domainUsersCanEdit', False) + + print(f"Sheet/Tab: {title}") + print(f"Description: {pr.get('description', '(none)')}") + print(f"Scope: {describe_range(pr.get('range'))}") + print(f"Warning only: {pr.get('warningOnly', False)}") + print(f"Editor list: {editor_users if editor_users else '(empty/not visible)'}") + print(f"Domain can edit: {domain_can_edit}") + + if pr.get('warningOnly', False): + print(" -> Warning-only: this protection would NOT block the service account's writes.") + elif sa_email in editor_users: + print(f" -> Service account IS listed as an allowed editor for this protection.") + else: + print(f" -> Service account is NOT listed. THIS PROTECTION IS LIKELY BLOCKING WRITES.") + print() + + if not found_any: + print("No protected ranges found on any sheet/tab in this spreadsheet.") + print("(If writes are still failing with a 400 protection error, double check") + print(" you're pointing at the correct spreadsheet ID / tab name, or that the") + print(" service account even has enough access to see the protection metadata.)") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scan-batch-dir b/scan-batch-dir index d212480..e84f3e1 100755 --- a/scan-batch-dir +++ b/scan-batch-dir @@ -119,6 +119,117 @@ def connect_to_google_sheet(credentials_file: str): except Exception as e: raise Exception(f"Failed to create Google Sheets service: {str(e)}") +def get_service_account_email(credentials_file: str) -> str: + """ + Reads the 'client_email' field out of a Google service account + credentials JSON file. + + Args: + credentials_file (str): Path to the Google service account credentials file. + + Returns: + str or None: The service account email if found, else None. + """ + try: + with open(credentials_file, "r") as f: + data = json.load(f) + return data.get("client_email") + except Exception as e: + logger.error(f"get_service_account_email - Failed to read credentials file: {str(e)}") + return None + + +def check_sheet_protections(spreadsheet_id: str, credentials_file: str) -> tuple: + """ + Checks all protected ranges/sheets in a Google Spreadsheet and determines + whether the service account associated with credentials_file is allowed + to edit within each protected range. + + Args: + spreadsheet_id (str): The ID of the Google Sheet. + credentials_file (str): Path to the Google service account credentials file. + + Returns: + tuple: (can_edit_all, blocking_protections, message) + can_edit_all (bool): True if no protections block the service account, + False if one or more protections would block writes. + blocking_protections (list of dict): Details of each protection that + the service account is NOT permitted to bypass. Each dict has + keys: sheet, description, range, warning_only. + message (str): Human-readable summary. + """ + try: + sa_email = get_service_account_email(credentials_file) + if not sa_email: + msg = "check_sheet_protections - Could not determine service account email from credentials file." + logger.error(msg) + return False, [], msg + + service = connect_to_google_sheet(credentials_file) + + result = service.spreadsheets().get( + spreadsheetId=spreadsheet_id, + fields="sheets(properties(sheetId,title),protectedRanges)" + ).execute() + + blocking_protections = [] + + for sheet in result.get('sheets', []): + title = sheet['properties']['title'] + for pr in sheet.get('protectedRanges', []): + warning_only = pr.get('warningOnly', False) + + # Warning-only protections do not block API writes. + if warning_only: + continue + + editors = pr.get('editors', {}) + editor_users = editors.get('users', []) + domain_can_edit = editors.get('domainUsersCanEdit', False) + + # If the service account is explicitly listed, it can edit. + if sa_email in editor_users: + continue + + # If domain-wide editing is allowed, and the service account + # belongs to that domain, it can edit. We can't reliably + # verify domain membership here, so we flag it as a caveat + # rather than assuming it's safe. + if domain_can_edit: + logger.warning( + f"check_sheet_protections - Sheet '{title}' protection allows " + f"domain editors; unable to verify if '{sa_email}' is in-domain." + ) + + # Otherwise, this protection blocks the service account. + blocking_protections.append({ + 'sheet': title, + 'description': pr.get('description', ''), + 'range': pr.get('range', {}), + 'warning_only': warning_only, + }) + + if blocking_protections: + msg = ( + f"check_sheet_protections - Service account '{sa_email}' is blocked by " + f"{len(blocking_protections)} protected range(s): " + + ", ".join( + f"[{p['sheet']}: {p['description'] or 'unnamed'}]" + for p in blocking_protections + ) + ) + logger.error(msg) + return False, blocking_protections, msg + else: + msg = f"check_sheet_protections - Service account '{sa_email}' can edit all ranges checked." + logger.info(msg) + return True, [], msg + + except Exception as e: + msg = f"check_sheet_protections - An error occurred: {str(e)}" + logger.error(msg) + return False, [], msg + def pad_rows(data: list) -> tuple: """ Pads each row in the data to ensure all rows have the same length. @@ -1534,6 +1645,17 @@ def main(): print(f"Reading Google Sheet: {google_sheet_id},{google_sheet_name}") df = read_google_sheet(google_sheet_id, google_sheet_name, google_credentials) + # Check that the service account can actually write to this sheet before + # doing all the work of scanning the directory. + can_edit, blocking, msg = check_sheet_protections(google_sheet_id, google_credentials) + if not can_edit: + logger.error(f"Aborting: {msg}") + print(f"Error: The service account cannot edit this sheet due to protected ranges:") + for p in blocking: + print(f" - Sheet '{p['sheet']}': {p['description'] or '(no description)'} — range: {p['range']}") + print("Contact the spreadsheet owner to remove protection or add the service account as an editor.") + sys.exit(1) + # Fail fast if the sheet could not be read, rather than crashing later # with an unhelpful AttributeError on a None DataFrame. if df is None: @@ -1544,7 +1666,13 @@ def main(): # Ensure all required_columns exist. for col in required_columns: df = add_column(df,col) - + + dupe_cols = df.columns[df.columns.duplicated()].tolist() + if dupe_cols: + logger.error(f"Duplicate columns detected in sheet: {dupe_cols}") + print(f"Duplicate columns detected in sheet: {dupe_cols}") + sys.exit(1) + # Fetch taxonomy terms once and cache as a name->tid dict. logger.info(f"Fetching Islandora Models taxonomy terms.") print(f"Fetching Islandora Models taxonomy terms.")