0
# HTTP Integration
1

2
Requests authentication plugin for seamless integration with HTTP clients, automatically handling Cognito token management and authentication headers. The RequestsSrpAuth class provides automatic authentication for HTTP requests using Cognito tokens.
3

4
## Capabilities
5

6
### RequestsSrpAuth Class
7

8
A Requests authentication plugin that automatically populates Authorization headers with Cognito tokens and handles token renewal.
9

10
```python { .api }
11
class RequestsSrpAuth(requests.auth.AuthBase):
12
    """
13
    Requests Auth Plugin for automatic Cognito token authentication.
14
    
15
    Automatically handles:
16
    - Initial authentication if no tokens present
17
    - Token expiration checking and renewal
18
    - Authorization header population
19
    - Token type selection (ID or Access token)
20
    """
21
    
22
    def __init__(self, username: str = None, password: str = None, 
23
                 user_pool_id: str = None, user_pool_region: str = None,
24
                 client_id: str = None, cognito: Cognito = None,
25
                 http_header: str = "Authorization", http_header_prefix: str = "Bearer ",
26
                 auth_token_type: TokenType = TokenType.ACCESS_TOKEN,
27
                 boto3_client_kwargs: dict = None):
28
        """
29
        Initialize RequestsSrpAuth plugin.
30
        
31
        Args:
32
            username (str, optional): Cognito username (required if cognito not provided)
33
            password (str, optional): User password (required if cognito not provided)
34
            user_pool_id (str, optional): User pool ID (required if cognito not provided)
35
            user_pool_region (str, optional): AWS region (required if cognito not provided)
36
            client_id (str, optional): Client ID (required if cognito not provided)
37
            cognito (Cognito, optional): Pre-configured Cognito instance
38
            http_header (str): HTTP header name for token (default: "Authorization")  
39
            http_header_prefix (str): Token prefix (default: "Bearer ")
40
            auth_token_type (TokenType): Token type to use (ACCESS_TOKEN or ID_TOKEN)
41
            boto3_client_kwargs (dict, optional): Additional boto3 client arguments
42
            
43
        Usage Modes:
44
            1. Individual parameters: Provide username, password, user_pool_id, etc.
45
            2. Pre-configured Cognito: Provide existing Cognito instance
46
            
47
        Note:
48
            Either provide individual authentication parameters OR a Cognito instance,
49
            not both.
50
        """
51
```
52

53
**Usage Example:**
54

55
```python
56
import requests
57
from pycognito.utils import RequestsSrpAuth, TokenType
58

59
# Method 1: Using individual parameters
60
auth = RequestsSrpAuth(
61
    username='user@example.com',
62
    password='password',
63
    user_pool_id='us-east-1_example123',
64
    client_id='your-client-id',
65
    user_pool_region='us-east-1'
66
)
67

68
# Method 2: Using pre-configured Cognito instance
69
from pycognito import Cognito
70

71
cognito_instance = Cognito(
72
    user_pool_id='us-east-1_example123',
73
    client_id='your-client-id',
74
    username='user@example.com'
75
)
76
cognito_instance.authenticate(password='password')
77

78
auth = RequestsSrpAuth(cognito=cognito_instance)
79

80
# Method 3: Custom configuration 
81
auth = RequestsSrpAuth(
82
    username='user@example.com',
83
    password='password',
84
    user_pool_id='us-east-1_example123',
85
    client_id='your-client-id',
86
    user_pool_region='us-east-1',
87
    http_header='X-Auth-Token',
88
    http_header_prefix='Token ',
89
    auth_token_type=TokenType.ID_TOKEN
90
)
91
```
92

93
### Automatic Authentication
94

95
The auth plugin automatically handles authentication and token management during HTTP requests.
96

97
```python { .api }
98
def __call__(self, request: requests.Request) -> requests.Request:
99
    """
100
    Process request and add authentication header.
101
    
102
    Args:
103
        request (requests.Request): HTTP request to authenticate
104
        
105
    Returns:
106
        requests.Request: Request with authentication header added
107
        
108
    Automatic Actions:
109
        1. Check if user is authenticated (has access token)
110
        2. If not authenticated, perform initial authentication
111
        3. Check if current token is expired
112
        4. If expired, automatically renew using refresh token
113
        5. Add appropriate token to specified HTTP header
114
        6. Return modified request
115
        
116
    Header Format:
117
        {http_header}: {http_header_prefix}{token}
118
        
119
    Default:
120
        Authorization: Bearer {access_token}
121
    """
122
```
123

124
**Usage Example:**
125

126
```python
127
import requests
128
from pycognito.utils import RequestsSrpAuth
129

130
# Set up authentication
131
auth = RequestsSrpAuth(
132
    username='user@example.com',
133
    password='password',
134
    user_pool_id='us-east-1_example123',
135
    client_id='your-client-id',
136
    user_pool_region='us-east-1'
137
)
138

139
# Make authenticated requests - token handling is automatic
140
response = requests.get('https://api.example.com/protected', auth=auth)
141

142
if response.status_code == 200:
143
    print("Request successful!")
144
    print(response.json())
145
else:
146
    print(f"Request failed: {response.status_code}")
147
```
148

149
### Token Type Selection
150

151
Choose between ID tokens and access tokens for different use cases.
152

153
```python { .api }
154
class TokenType(str, Enum):
155
    """Token types for authentication headers."""
156
    ID_TOKEN = "id_token"
157
    ACCESS_TOKEN = "access_token"
158
```
159

160
**Usage Example:**
161

162
```python
163
from pycognito.utils import RequestsSrpAuth, TokenType
164

165
# Use access token (default - for API authorization)
166
api_auth = RequestsSrpAuth(
167
    username='user@example.com',
168
    password='password',
169
    user_pool_id='us-east-1_example123',
170
    client_id='your-client-id',
171
    auth_token_type=TokenType.ACCESS_TOKEN
172
)
173

174
# Use ID token (for user identity/profile information)
175
identity_auth = RequestsSrpAuth(
176
    username='user@example.com',
177
    password='password',
178
    user_pool_id='us-east-1_example123',
179
    client_id='your-client-id',
180
    auth_token_type=TokenType.ID_TOKEN
181
)
182

183
# API calls
184
api_response = requests.get('https://api.example.com/data', auth=api_auth)
185

186
# Identity/profile calls  
187
profile_response = requests.get('https://api.example.com/profile', auth=identity_auth)
188
```
189

190
## Usage Patterns
191

192
### Basic API Client with Authentication
193

194
```python
195
import requests
196
from pycognito.utils import RequestsSrpAuth
197

198
class APIClient:
199
    """API client with automatic Cognito authentication."""
200
    
201
    def __init__(self, base_url, username, password, user_pool_id, client_id, region):
202
        self.base_url = base_url
203
        
204
        # Set up authentication
205
        self.auth = RequestsSrpAuth(
206
            username=username,
207
            password=password,
208
            user_pool_id=user_pool_id,
209
            client_id=client_id,
210
            user_pool_region=region
211
        )
212
    
213
    def get(self, endpoint, **kwargs):
214
        """Make authenticated GET request."""
215
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
216
        return requests.get(url, auth=self.auth, **kwargs)
217
    
218
    def post(self, endpoint, data=None, json=None, **kwargs):
219
        """Make authenticated POST request."""
220
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
221
        return requests.post(url, data=data, json=json, auth=self.auth, **kwargs)
222
    
223
    def put(self, endpoint, data=None, json=None, **kwargs):
224
        """Make authenticated PUT request."""
225
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
226
        return requests.put(url, data=data, json=json, auth=self.auth, **kwargs)
227
    
228
    def delete(self, endpoint, **kwargs):
229
        """Make authenticated DELETE request."""
230
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
231
        return requests.delete(url, auth=self.auth, **kwargs)
232

233
# Usage
234
client = APIClient(
235
    base_url='https://api.example.com',
236
    username='user@example.com',
237
    password='password',
238
    user_pool_id='us-east-1_example123',
239
    client_id='your-client-id',
240
    region='us-east-1'
241
)
242

243
# All requests automatically include authentication
244
users = client.get('/users')
245
user_data = client.post('/users', json={'name': 'John Doe'})
246
```
247

248
### Session-Based HTTP Client
249

250
```python
251
import requests
252
from pycognito.utils import RequestsSrpAuth
253

254
def create_authenticated_session(username, password, user_pool_id, client_id, region):
255
    """Create requests Session with automatic Cognito authentication."""
256
    
257
    # Create session
258
    session = requests.Session()
259
    
260
    # Set up authentication
261
    session.auth = RequestsSrpAuth(
262
        username=username,
263
        password=password,
264
        user_pool_id=user_pool_id,
265
        client_id=client_id,
266
        user_pool_region=region
267
    )
268
    
269
    # Optional: Set common headers
270
    session.headers.update({
271
        'Content-Type': 'application/json',
272
        'User-Agent': 'MyApp/1.0'
273
    })
274
    
275
    return session
276

277
# Usage
278
session = create_authenticated_session(
279
    username='user@example.com',
280
    password='password',
281
    user_pool_id='us-east-1_example123',
282
    client_id='your-client-id',
283
    region='us-east-1'
284
)
285

286
# All requests in this session are automatically authenticated
287
response1 = session.get('https://api.example.com/endpoint1')
288
response2 = session.post('https://api.example.com/endpoint2', json={'data': 'value'})
289
response3 = session.put('https://api.example.com/endpoint3', json={'update': 'value'})
290
```
291

292
### Custom Header Authentication
293

294
```python
295
from pycognito.utils import RequestsSrpAuth, TokenType
296

297
# Custom header for legacy APIs
298
custom_auth = RequestsSrpAuth(
299
    username='user@example.com',
300
    password='password',
301
    user_pool_id='us-east-1_example123',
302
    client_id='your-client-id',
303
    user_pool_region='us-east-1',
304
    http_header='X-API-Key',
305
    http_header_prefix='',  # No prefix
306
    auth_token_type=TokenType.ACCESS_TOKEN
307
)
308

309
# API expects: X-API-Key: {access_token}
310
response = requests.get('https://legacy-api.example.com/data', auth=custom_auth)
311

312
# JWT Bearer token for modern APIs
313
jwt_auth = RequestsSrpAuth(
314
    username='user@example.com', 
315
    password='password',
316
    user_pool_id='us-east-1_example123',
317
    client_id='your-client-id',
318
    user_pool_region='us-east-1',
319
    http_header='Authorization',
320
    http_header_prefix='JWT ',
321
    auth_token_type=TokenType.ID_TOKEN
322
)
323

324
# API expects: Authorization: JWT {id_token}
325
response = requests.get('https://jwt-api.example.com/profile', auth=jwt_auth)
326
```
327

328
### Error Handling and Retry Logic
329

330
```python
331
import requests
332
from requests.adapters import HTTPAdapter
333
from requests.packages.urllib3.util.retry import Retry
334
from pycognito.utils import RequestsSrpAuth
335

336
class RobustAPIClient:
337
    """API client with retry logic and error handling."""
338
    
339
    def __init__(self, base_url, username, password, user_pool_id, client_id, region):
340
        self.base_url = base_url
341
        
342
        # Create session with retry strategy
343
        self.session = requests.Session()
344
        
345
        # Configure retry strategy
346
        retry_strategy = Retry(
347
            total=3,
348
            backoff_factor=1,
349
            status_forcelist=[429, 500, 502, 503, 504],
350
        )
351
        
352
        adapter = HTTPAdapter(max_retries=retry_strategy)
353
        self.session.mount("http://", adapter)
354
        self.session.mount("https://", adapter)
355
        
356
        # Set up authentication
357
        self.session.auth = RequestsSrpAuth(
358
            username=username,
359
            password=password,
360
            user_pool_id=user_pool_id,
361
            client_id=client_id,
362
            user_pool_region=region
363
        )
364
    
365
    def make_request(self, method, endpoint, **kwargs):
366
        """Make request with comprehensive error handling."""
367
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
368
        
369
        try:
370
            response = self.session.request(method, url, **kwargs)
371
            response.raise_for_status()
372
            return response
373
            
374
        except requests.exceptions.HTTPError as e:
375
            if e.response.status_code == 401:
376
                print("Authentication failed - check credentials")
377
            elif e.response.status_code == 403:
378
                print("Access forbidden - check permissions")
379
            else:
380
                print(f"HTTP error: {e}")
381
            raise
382
            
383
        except requests.exceptions.ConnectionError:
384
            print("Connection error - check network")
385
            raise
386
            
387
        except requests.exceptions.Timeout:
388
            print("Request timeout - server may be overloaded")
389
            raise
390
            
391
        except requests.exceptions.RequestException as e:
392
            print(f"Request failed: {e}")
393
            raise
394

395
# Usage
396
client = RobustAPIClient(
397
    base_url='https://api.example.com',
398
    username='user@example.com',
399
    password='password',
400
    user_pool_id='us-east-1_example123',
401
    client_id='your-client-id',
402
    region='us-east-1'
403
)
404

405
try:
406
    response = client.make_request('GET', '/protected-data')
407
    data = response.json()
408
    print(f"Success: {data}")
409
except Exception as e:
410
    print(f"Request failed: {e}")
411
```
412

413
### Multi-Environment Configuration
414

415
```python
416
from pycognito.utils import RequestsSrpAuth
417
import os
418

419
class MultiEnvironmentAuth:
420
    """Authentication manager for multiple environments."""
421
    
422
    ENVIRONMENTS = {
423
        'dev': {
424
            'user_pool_id': 'us-east-1_dev123',
425
            'client_id': 'dev-client-id',
426
            'region': 'us-east-1',
427
            'base_url': 'https://dev-api.example.com'
428
        },
429
        'staging': {
430
            'user_pool_id': 'us-east-1_staging456',
431
            'client_id': 'staging-client-id', 
432
            'region': 'us-east-1',
433
            'base_url': 'https://staging-api.example.com'
434
        },
435
        'prod': {
436
            'user_pool_id': 'us-east-1_prod789',
437
            'client_id': 'prod-client-id',
438
            'region': 'us-east-1',
439
            'base_url': 'https://api.example.com'
440
        }
441
    }
442
    
443
    def __init__(self, environment='dev'):
444
        self.env = environment
445
        self.config = self.ENVIRONMENTS[environment]
446
    
447
    def create_auth(self, username, password):
448
        """Create auth for specified environment."""
449
        return RequestsSrpAuth(
450
            username=username,
451
            password=password,
452
            user_pool_id=self.config['user_pool_id'],
453
            client_id=self.config['client_id'],
454
            user_pool_region=self.config['region']
455
        )
456
    
457
    def make_request(self, method, endpoint, username, password, **kwargs):
458
        """Make authenticated request to environment."""
459
        auth = self.create_auth(username, password)
460
        url = f"{self.config['base_url']}/{endpoint.lstrip('/')}"
461
        
462
        return requests.request(method, url, auth=auth, **kwargs)
463

464
# Usage
465
# Development environment
466
dev_auth = MultiEnvironmentAuth('dev')
467
dev_response = dev_auth.make_request(
468
    'GET', '/users',
469
    username='dev-user@example.com',
470
    password='dev-password'
471
)
472

473
# Production environment  
474
prod_auth = MultiEnvironmentAuth('prod')
475
prod_response = prod_auth.make_request(
476
    'GET', '/users',
477
    username='prod-user@example.com', 
478
    password='prod-password'
479
)
480

481
# Environment from environment variable
482
env = os.getenv('API_ENVIRONMENT', 'dev')
483
auth_manager = MultiEnvironmentAuth(env)
484
```
485

486
### Async HTTP Integration
487

488
```python
489
import asyncio
490
import aiohttp
491
from pycognito.utils import RequestsSrpAuth, TokenType
492

493
class AsyncCognitoAuth:
494
    """Async-compatible Cognito authentication."""
495
    
496
    def __init__(self, username, password, user_pool_id, client_id, region):
497
        # Initialize sync auth to get tokens
498
        self.sync_auth = RequestsSrpAuth(
499
            username=username,
500
            password=password,
501
            user_pool_id=user_pool_id,
502
            client_id=client_id,
503
            user_pool_region=region,
504
            auth_token_type=TokenType.ACCESS_TOKEN
505
        )
506
        
507
        # Perform initial authentication
508
        import requests
509
        dummy_request = requests.Request('GET', 'http://example.com')
510
        self.sync_auth(dummy_request)  # This triggers authentication
511
    
512
    def get_headers(self):
513
        """Get authentication headers for async requests."""
514
        token = getattr(self.sync_auth.cognito_client, self.sync_auth.token_type.value)
515
        return {
516
            self.sync_auth.http_header: f"{self.sync_auth.http_header_prefix}{token}"
517
        }
518
    
519
    def refresh_if_needed(self):
520
        """Refresh token if needed (sync operation)."""
521
        self.sync_auth.cognito_client.check_token(renew=True)
522

523
async def async_api_client():
524
    """Example async API client with Cognito authentication."""
525
    
526
    # Set up authentication
527
    auth = AsyncCognitoAuth(
528
        username='user@example.com',
529
        password='password',
530
        user_pool_id='us-east-1_example123',
531
        client_id='your-client-id',
532
        region='us-east-1'
533
    )
534
    
535
    # Make async requests
536
    async with aiohttp.ClientSession() as session:
537
        
538
        # Refresh token if needed before making requests
539
        auth.refresh_if_needed()
540
        headers = auth.get_headers()
541
        
542
        # Make multiple concurrent requests
543
        tasks = []
544
        for i in range(5):
545
            task = session.get(
546
                f'https://api.example.com/data/{i}',
547
                headers=headers
548
            )
549
            tasks.append(task)
550
        
551
        responses = await asyncio.gather(*tasks)
552
        
553
        # Process responses
554
        for i, response in enumerate(responses):
555
            if response.status == 200:
556
                data = await response.json()
557
                print(f"Request {i} successful: {data}")
558
            else:
559
                print(f"Request {i} failed: {response.status}")
560

561
# Usage
562
asyncio.run(async_api_client())
563
```