Impersonate users

1
import requests
2
import jwt
3
import json
4
5
requests.packages.urllib3.disable_warnings()
6
7
# provide your client ID, client secret,
8
# and base URL for Sigma REST API calls
9
# as environment variables
10
client_id=""
11
client_secret=""
12
url = ""
13
# tenant_region_url = ""
14
15
# Get the admin access token to use as the actor token
16
def get_actor_token():
17
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
18
    payload = {
19
        "grant_type": "client_credentials",
20
        "client_id": client_id,
21
        "client_secret": client_secret
22
    }
23
24
    try:
25
        response = requests.post(f"{url}/v2/auth/token", headers=headers, data=payload, verify=False)
26
        response.raise_for_status()
27
        result = response.json()
28
        print(f"Generated actor token: {result}")
29
        return result["access_token"]
30
    
31
    except requests.exceptions.RequestException as e:
32
        print(f"Error getting actor token: {e}")
33
        if hasattr(e, 'response') and e.response is not None:
34
            print(f"Response status: {e.response.status_code}")
35
            print(f"Response body: {e.response.text}")
36
        raise
37
38
# Generate the self-signed JWT token
39
# Provide the email address of the user to impersonate
40
def generate_subject_token():
41
    jwt_payload = {
42
        # to impersonate a user, provide the sub claim
43
        "sub": "<email address of user to impersonate>",
44
        # to impersonate a user in a tenant organization, also provide the tenant claim
45
        # "tenant": "<organization ID of a tenant organization>",
46
        # to perform actions in a tenant organization as the TenantAdminRobot user, only provide the tenant claim
47
    }
48
49
    jwt_header = {
50
        "alg": "HS256",
51
        "kid": client_id,
52
        "typ": "JWT"
53
    }
54
55
    try:
56
        jwt_token = jwt.encode(
57
            jwt_payload,
58
            client_secret,
59
            algorithm="HS256",
60
            headers=jwt_header
61
        )
62
        print(f"Generated subject token: {jwt_token[:50]}...")
63
        return jwt_token
64
    except Exception as e:
65
        print(f"Error generating subject token: {e}")
66
        raise
67
68
def exchange_token(subject_token, actor_token):
69
70
    # Get impersonation token
71
    try:
72
        headers = {
73
            "Content-Type": "application/x-www-form-urlencoded",
74
            "Accept": "application/json",
75
            "User-Agent": "Python-OAuth-Client/1.0",
76
            "Authorization": f"Bearer {actor_token}"
77
        }
78
79
        payload = {
80
            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
81
            "subject_token": subject_token,
82
            "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
83
            "actor_token": actor_token,
84
            "actor_token_type": "urn:ietf:params:oauth:token-type:access_token",
85
        }
86
87
        print(f"Token exchange payload: {json.dumps({k: v[:50] + '...' if len(str(v)) > 50 else v for k, v in payload.items()}, indent=2)}")
88
89
        response = requests.post(
90
            f"{url}/v2/auth/token",
91
            headers=headers,
92
            data=payload,
93
            verify=False,
94
            timeout=30
95
        )
96
97
        print(f"Token exchange response status: {response.status_code}")
98
        print(f"Token exchange response headers: {dict(response.headers)}")
99
        print(f"Token exchange response body: {response.text}")
100
101
        response.raise_for_status()
102
        return response.json()
103
104
    except requests.exceptions.RequestException as e:
105
        print(f"Error during token exchange: {e}")
106
        if hasattr(e, 'response') and e.response is not None:
107
            print(f"Response status: {e.response.status_code}")
108
            print(f"Response body: {e.response.text}")
109
        raise
110
    except Exception as e:
111
        print(f"Unexpected error during token exchange: {e}")
112
        raise
113
114
# Retrieve the ID of the impersonated user
115
def get_current_user(token):
116
    headers = {
117
        "Authorization": f"Bearer {token}"
118
    }
119
120
    try:
121
        response = requests.get(
122
            f"{url}/v2/whoami",
123
            headers=headers,
124
            verify=False,
125
            timeout=30
126
        )
127
128
        print(f"Whoami response status: {response.status_code}")
129
130
        response.raise_for_status()
131
        return response.json()
132
133
    except requests.exceptions.RequestException as e:
134
        print(f"Error retrieving current user: {e}")
135
        if hasattr(e, 'response') and e.response is not None:
136
            print(f"Response status: {e.response.status_code}")
137
            print(f"Response body: {e.response.text}")
138
        raise
139
    except Exception as e:
140
        print(f"Unexpected error retrieving current user: {e}")
141
        raise
142
143
# List workbook files accessible by current user
144
def list_files_for_user(token):
145
    headers = {
146
        "Authorization": f"Bearer {token}"
147
    }
148
149
    try:
150
        response = requests.get(
151
            f"{url}/v2/files?typeFilters=workbook",
152
            # for tenant impersonation for tenants in other regions, instead use:
153
            # f"{tenant_region_url}/v2/files?typeFilters=workbook",
154
            headers=headers,
155
            verify=False,
156
            timeout=30
157
        )
158
159
        print(f"List workbook files response status: {response.status_code}")
160
161
        response.raise_for_status()
162
        return response.json()
163
164
    except requests.exceptions.RequestException as e:
165
        print(f"Error listing workbook files for the current user: {e}")
166
        if hasattr(e, 'response') and e.response is not None:
167
            print(f"Response status: {e.response.status_code}")
168
            print(f"Response body: {e.response.text}")
169
        raise
170
    except Exception as e:
171
        print(f"Unexpected error listing workbook files for the current user: {e}")
172
        raise
173
174
# Run all defined functions as a test
175
def main():
176
    print("=== OAuth Token Exchange Test ===")
177
178
    try:
179
        print("\n1. Generating subject token...")
180
        subject_token = generate_subject_token()
181
        print(f"Subject token: {subject_token[:100]}...")
182
183
        print("\n2. Getting actor token...")
184
        actor_token = get_actor_token()
185
        print(f"Actor token: {actor_token[:100]}...")
186
187
        print("\n3. Exchanging tokens to impersonate user...")
188
        result = exchange_token(subject_token, actor_token)
189
        print(f"Token impersonation result: {json.dumps(result, indent=2)}")
190
191
        print("\n4. Get current user with actor token...")
192
        whoami_a = get_current_user(actor_token)
193
        print(f"Current user: {whoami_a}")
194
195
        print("\n5. Get current user with impersonation token...")
196
        whoami_b = get_current_user(result["access_token"])
197
        print(f"Current user: {whoami_b}")
198
199
        print("\n6. Listing workbook files for user with actor token...")
200
        workbooks = list_files_for_user(actor_token)
201
        print(f"Workbooks accessible by current user: {workbooks}")
202
203
        print("\n7. Listing workbook files for user with impersonation token...")
204
        workbooks = list_files_for_user(result["access_token"])
205
        print(f"Workbooks accessible by current user: {workbooks}")
206
207
    except Exception as e:
208
        print(f"Script failed with error: {e}")
209
        import traceback
210
        traceback.print_exc()
211
212
if __name__ == "__main__":
213
    main()