grant_page_edit.py

gistfile1.txt

#!/usr/bin/env python3
"""
Script to grant edit permissions to a user on a specific Confluence page.

Usage: python grant_page_edit.py <page_id> <user_email> [--notify]
"""

import sys
import requests
import credentials

USAGE = """
Usage: python grant_page_edit.py <page_id> <user_email> [--notify]
  page_id: The Confluence page ID
  user_email: Email address of the user to grant edit permissions to
  --notify: (Optional) Send notification to the user about the permission change
"""

BASE_URL = f"{credentials.HOSTNAME}"
JIRA_API_URL = f"{BASE_URL}/rest/api"
CONFLUENCE_API_URL = f"{BASE_URL}/wiki/rest/api"
GRAPHQL_URL = f"{credentials.HOSTNAME}/cgraphql"


def get_account_id_from_email(email):
    """Look up Atlassian account ID from email address."""
    url = f"{JIRA_API_URL}/3/user/search"
    params = {"query": email}
    headers = {"Accept": "application/json"}
    
    resp = requests.get(
        url,
        params=params,
        auth=(credentials.USERNAME, credentials.TOKEN),
        headers=headers
    )
    resp.raise_for_status()
    
    users = resp.json()
    if not users:
        print(f"ERROR: No user found with email '{email}'")
        sys.exit(1)
    
    # Find exact match
    for user in users:
        if user.get("emailAddress", "").lower() == email.lower():
            account_id = user.get("accountId")
            display_name = user.get("displayName")
            print(f"Found user: {display_name} ({email}) -> Account ID: {account_id}")
            return account_id
    
    print(f"ERROR: No exact match found for email '{email}'")
    sys.exit(1)


def get_current_page_restrictions(page_id):
    """Get current page restrictions using REST API."""
    url = f"{CONFLUENCE_API_URL}/content/{page_id}/restriction"
    headers = {"Accept": "application/json"}
    
    resp = requests.get(
        url,
        auth=(credentials.USERNAME, credentials.TOKEN),
        headers=headers
    )
    resp.raise_for_status()
    
    data = resp.json()
    
    # Parse the REST API response format
    restrictions = {
        "read": {"group": [], "user": []},
        "update": {"group": [], "user": []}
    }
    
    for result in data.get("results", []):
        operation = result.get("operation")
        if operation not in ["read", "update"]:
            continue
        
        # Extract user account IDs
        user_results = result.get("restrictions", {}).get("user", {}).get("results", [])
        for user in user_results:
            account_id = user.get("accountId")
            if account_id:
                restrictions[operation]["user"].append({"id": account_id})
        
        # Extract group IDs (if any)
        group_results = result.get("restrictions", {}).get("group", {}).get("results", [])
        for group in group_results:
            group_id = group.get("id")  # or group.get("name") depending on API
            if group_id:
                restrictions[operation]["group"].append({"id": group_id})
    
    return restrictions


def grant_edit_permission(page_id, account_id, send_notification=False):
    """Grant edit permission to a user on a page using GraphQL."""
    print(f"\nChecking permissions for user {account_id} on page {page_id}...")
    
    # Get current restrictions
    current_restrictions = get_current_page_restrictions(page_id)
    
    # Extract current update users
    update_users = [u["id"] for u in current_restrictions.get("update", {}).get("user", [])]
    
    # Check if user already has permissions
    if account_id in update_users:
        print(f"✓ User already has update permissions - no action needed")
        return True
    
    # User doesn't have permissions, proceed to grant them
    print(f"User does not have update permissions - granting access...")
    print(f"Send notification: {send_notification}")
    
    # Extract all current users and groups
    read_groups = [g["id"] for g in current_restrictions.get("read", {}).get("group", [])]
    read_users = [u["id"] for u in current_restrictions.get("read", {}).get("user", [])]
    update_groups = [g["id"] for g in current_restrictions.get("update", {}).get("group", [])]
    
    # Add the new user
    update_users.append(account_id)
    
    # Prepare the mutation
    headers = {
        "Content-Type": "application/json",
        "X-APOLLO-OPERATION-NAME": "ControllerDirectRestrictionsMutation"
    }
    
    query = """mutation ControllerDirectRestrictionsMutation($pageId: ID!, $restrictions: PageRestrictionsInput) {
  updatePage(input: {pageId: $pageId, restrictions: $restrictions}) {
    page {
      id
      __typename
    }
    __typename
  }
}"""
    
    # Build restrictions object
    restrictions = {
        "read": {
            "group": [{"id": gid} for gid in read_groups],
            "user": [{"id": uid} for uid in read_users]
        },
        "update": {
            "group": [{"id": gid} for gid in update_groups],
            "user": [{"id": uid} for uid in update_users]
        },
        "includeInvites": send_notification
    }
    
    payload = [{
        "operationName": "ControllerDirectRestrictionsMutation",
        "variables": {
            "pageId": page_id,
            "restrictions": restrictions
        },
        "query": query
    }]
    
    resp = requests.post(
        f"{GRAPHQL_URL}?q=ControllerDirectRestrictionsMutation",
        auth=(credentials.USERNAME, credentials.TOKEN),
        headers=headers,
        json=payload
    )
    
    if resp.status_code in [200, 201]:
        data = resp.json()
        if isinstance(data, list) and len(data) > 0:
            errors = data[0].get("errors")
            if errors:
                print(f"✗ Failed to grant permissions: {errors}")
                return False
            else:
                print(f"✓ Successfully granted edit permissions")
                return True
    else:
        print(f"✗ Failed to grant permissions: {resp.status_code} - {resp.text}")
        return False


def main():
    if len(sys.argv) < 3:
        print(USAGE)
        sys.exit(1)
    
    page_id = sys.argv[1]
    user_email = sys.argv[2]
    send_notification = '--notify' in sys.argv
    
    print("=" * 80)
    print(f"Granting edit permissions on Confluence page")
    print("=" * 80)
    print(f"Page ID: {page_id}")
    print(f"User Email: {user_email}")
    print(f"Send Notification: {'Yes' if send_notification else 'No'}")
    print()
    
    # Look up user account ID
    account_id = get_account_id_from_email(user_email)
    
    # Grant edit permission
    success = grant_edit_permission(page_id, account_id, send_notification)
    
    if success:
        print("\n" + "=" * 80)
        print("✓ Edit permissions granted successfully!")
        print("=" * 80)
    else:
        print("\n" + "=" * 80)
        print("✗ Failed to grant edit permissions")
        print("=" * 80)
        sys.exit(1)


if __name__ == '__main__':
    main()