BITWARDEN_EVENTS

  { 
  
 "Version" 
 : 
  
 "2012-10-17" 
 , 
  
 "Statement" 
 : 
  
 [ 
  
 { 
  
 "Sid" 
 : 
  
 "AllowPutBitwardenObjects" 
 , 
  
 "Effect" 
 : 
  
 "Allow" 
 , 
  
 "Action" 
 : 
  
 "s3:PutObject" 
 , 
  
 "Resource" 
 : 
  
 "arn:aws:s3:::bitwarden-events/*" 
  
 }, 
  
 { 
  
 "Sid" 
 : 
  
 "AllowGetStateObject" 
 , 
  
 "Effect" 
 : 
  
 "Allow" 
 , 
  
 "Action" 
 : 
  
 "s3:GetObject" 
 , 
  
 "Resource" 
 : 
  
 "arn:aws:s3:::bitwarden-events/bitwarden/events/state.json" 
  
 } 
  
 ] 
 } 
 

  #!/usr/bin/env python3 
 import 
  
 os 
 , 
  
 json 
 , 
  
 time 
 , 
  
 urllib.parse 
 from 
  
 urllib.request 
  
 import 
 Request 
 , 
 urlopen 
 from 
  
 urllib.error 
  
 import 
 HTTPError 
 , 
 URLError 
 import 
  
 boto3 
 IDENTITY_URL 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "IDENTITY_URL" 
 , 
 "https://identity.bitwarden.com/connect/token" 
 ) 
 API_BASE 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "API_BASE" 
 , 
 "https://api.bitwarden.com" 
 ) 
 . 
 rstrip 
 ( 
 "/" 
 ) 
 CID 
 = 
 os 
 . 
 environ 
 [ 
 "BW_CLIENT_ID" 
 ] 
 # organization.ClientId 
 CSECRET 
 = 
 os 
 . 
 environ 
 [ 
 "BW_CLIENT_SECRET" 
 ] 
 # organization.ClientSecret 
 BUCKET 
 = 
 os 
 . 
 environ 
 [ 
 "S3_BUCKET" 
 ] 
 PREFIX 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "S3_PREFIX" 
 , 
 "bitwarden/events/" 
 ) 
 . 
 strip 
 ( 
 "/" 
 ) 
 STATE_KEY 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "STATE_KEY" 
 , 
 "bitwarden/events/state.json" 
 ) 
 MAX_PAGES 
 = 
 int 
 ( 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "MAX_PAGES" 
 , 
 "10" 
 )) 
 HEADERS_FORM 
 = 
 { 
 "Content-Type" 
 : 
 "application/x-www-form-urlencoded" 
 } 
 HEADERS_JSON 
 = 
 { 
 "Accept" 
 : 
 "application/json" 
 } 
 s3 
 = 
 boto3 
 . 
 client 
 ( 
 "s3" 
 ) 
 def 
  
 _read_state 
 (): 
 try 
 : 
 obj 
 = 
 s3 
 . 
 get_object 
 ( 
 Bucket 
 = 
 BUCKET 
 , 
 Key 
 = 
 STATE_KEY 
 ) 
 j 
 = 
 json 
 . 
 loads 
 ( 
 obj 
 [ 
 "Body" 
 ] 
 . 
 read 
 ()) 
 return 
 j 
 . 
 get 
 ( 
 "continuationToken" 
 ) 
 except 
 Exception 
 : 
 return 
 None 
 def 
  
 _write_state 
 ( 
 token 
 ): 
 body 
 = 
 json 
 . 
 dumps 
 ({ 
 "continuationToken" 
 : 
 token 
 }) 
 . 
 encode 
 ( 
 "utf-8" 
 ) 
 s3 
 . 
 put_object 
 ( 
 Bucket 
 = 
 BUCKET 
 , 
 Key 
 = 
 STATE_KEY 
 , 
 Body 
 = 
 body 
 , 
 ContentType 
 = 
 "application/json" 
 ) 
 def 
  
 _http 
 ( 
 req 
 : 
 Request 
 , 
 timeout 
 : 
 int 
 = 
 60 
 , 
 max_retries 
 : 
 int 
 = 
 5 
 ): 
 attempt 
 , 
 backoff 
 = 
 0 
 , 
 1.0 
 while 
 True 
 : 
 try 
 : 
 with 
 urlopen 
 ( 
 req 
 , 
 timeout 
 = 
 timeout 
 ) 
 as 
 r 
 : 
 return 
 json 
 . 
 loads 
 ( 
 r 
 . 
 read 
 () 
 . 
 decode 
 ( 
 "utf-8" 
 )) 
 except 
 HTTPError 
 as 
 e 
 : 
 # Retry on 429 and 5xx 
 if 
 ( 
 e 
 . 
 code 
 == 
 429 
 or 
 500 
< = 
 e 
 . 
 code 
< = 
 599 
 ) 
 and 
 attempt 
< max_retries 
 : 
 time 
 . 
 sleep 
 ( 
 backoff 
 ); 
 attempt 
 += 
 1 
 ; 
 backoff 
 *= 
 2 
 ; 
 continue 
 raise 
 except 
 URLError 
 : 
 if 
 attempt 
< max_retries 
 : 
 time 
 . 
 sleep 
 ( 
 backoff 
 ); 
 attempt 
 += 
 1 
 ; 
 backoff 
 *= 
 2 
 ; 
 continue 
 raise 
 def 
  
 _get_token 
 (): 
 body 
 = 
 urllib 
 . 
 parse 
 . 
 urlencode 
 ({ 
 "grant_type" 
 : 
 "client_credentials" 
 , 
 "scope" 
 : 
 "api.organization" 
 , 
 "client_id" 
 : 
 CID 
 , 
 "client_secret" 
 : 
 CSECRET 
 , 
 }) 
 . 
 encode 
 ( 
 "utf-8" 
 ) 
 req 
 = 
 Request 
 ( 
 IDENTITY_URL 
 , 
 data 
 = 
 body 
 , 
 method 
 = 
 "POST" 
 , 
 headers 
 = 
 HEADERS_FORM 
 ) 
 data 
 = 
 _http 
 ( 
 req 
 , 
 timeout 
 = 
 30 
 ) 
 return 
 data 
 [ 
 "access_token" 
 ], 
 int 
 ( 
 data 
 . 
 get 
 ( 
 "expires_in" 
 , 
 3600 
 )) 
 def 
  
 _fetch_events 
 ( 
 bearer 
 : 
 str 
 , 
 cont 
 : 
 str 
 | 
 None 
 ): 
 params 
 = 
 {} 
 if 
 cont 
 : 
 params 
 [ 
 "continuationToken" 
 ] 
 = 
 cont 
 qs 
 = 
 ( 
 "?" 
 + 
 urllib 
 . 
 parse 
 . 
 urlencode 
 ( 
 params 
 )) 
 if 
 params 
 else 
 "" 
 url 
 = 
 f 
 " 
 { 
 API_BASE 
 } 
 /public/events 
 { 
 qs 
 } 
 " 
 req 
 = 
 Request 
 ( 
 url 
 , 
 method 
 = 
 "GET" 
 , 
 headers 
 = 
 { 
 "Authorization" 
 : 
 f 
 "Bearer 
 { 
 bearer 
 } 
 " 
 , 
 ** 
 HEADERS_JSON 
 }) 
 return 
 _http 
 ( 
 req 
 , 
 timeout 
 = 
 60 
 ) 
 def 
  
 _write_events_jsonl 
 ( 
 events 
 : 
 list 
 , 
 run_ts_s 
 : 
 int 
 , 
 page_index 
 : 
 int 
 ) 
 - 
> str 
 : 
  
 """ 
 Write events in JSONL format (one JSON object per line). 
 Only writes if there are events to write. 
 Returns the S3 key of the written file. 
 """ 
 if 
 not 
 events 
 : 
 return 
 None 
 # Build JSONL content: one event per line 
 lines 
 = 
 [ 
 json 
 . 
 dumps 
 ( 
 event 
 , 
 separators 
 = 
 ( 
 "," 
 , 
 ":" 
 )) 
 for 
 event 
 in 
 events 
 ] 
 jsonl_content 
 = 
 " 
 \n 
 " 
 . 
 join 
 ( 
 lines 
 ) 
 + 
 " 
 \n 
 " 
 # JSONL format with trailing newline 
 # Generate unique filename with page number to avoid conflicts 
 key 
 = 
 f 
 " 
 { 
 PREFIX 
 } 
 / 
 { 
 time 
 . 
 strftime 
 ( 
 '%Y/%m/ 
 %d 
 /%H%M%S' 
 , 
  
 time 
 . 
 gmtime 
 ( 
 run_ts_s 
 )) 
 } 
 -page 
 { 
 page_index 
 : 
 05d 
 } 
 -bitwarden-events.jsonl" 
 s3 
 . 
 put_object 
 ( 
 Bucket 
 = 
 BUCKET 
 , 
 Key 
 = 
 key 
 , 
 Body 
 = 
 jsonl_content 
 . 
 encode 
 ( 
 "utf-8" 
 ), 
 ContentType 
 = 
 "application/x-ndjson" 
 , 
 # MIME type for JSONL 
 ) 
 return 
 key 
 def 
  
 lambda_handler 
 ( 
 event 
 = 
 None 
 , 
 context 
 = 
 None 
 ): 
 bearer 
 , 
 _ttl 
 = 
 _get_token 
 () 
 cont 
 = 
 _read_state 
 () 
 run_ts_s 
 = 
 int 
 ( 
 time 
 . 
 time 
 ()) 
 pages 
 = 
 0 
 total_events 
 = 
 0 
 written_files 
 = 
 [] 
 while 
 pages 
< MAX_PAGES 
 : 
 data 
 = 
 _fetch_events 
 ( 
 bearer 
 , 
 cont 
 ) 
 # Extract events array from API response 
 # API returns: {"object":"list", "data":[...], "continuationToken":"..."} 
 events 
 = 
 data 
 . 
 get 
 ( 
 "data" 
 , 
 []) 
 # Only write file if there are events 
 if 
 events 
 : 
 s3_key 
 = 
 _write_events_jsonl 
 ( 
 events 
 , 
 run_ts_s 
 , 
 pages 
 ) 
 if 
 s3_key 
 : 
 written_files 
 . 
 append 
 ( 
 s3_key 
 ) 
 total_events 
 += 
 len 
 ( 
 events 
 ) 
 pages 
 += 
 1 
 # Check for next page token 
 next_cont 
 = 
 data 
 . 
 get 
 ( 
 "continuationToken" 
 ) 
 if 
 next_cont 
 : 
 cont 
 = 
 next_cont 
 continue 
 else 
 : 
 # No more pages 
 break 
 # Save state only if there are more pages to continue in next run 
 # If we hit MAX_PAGES and there's still a continuation token, save it 
 # Otherwise, clear the state (set to None) 
 _write_state 
 ( 
 cont 
 if 
 pages 
> = 
 MAX_PAGES 
 and 
 cont 
 else 
 None 
 ) 
 return 
 { 
 "ok" 
 : 
 True 
 , 
 "pages" 
 : 
 pages 
 , 
 "total_events" 
 : 
 total_events 
 , 
 "files_written" 
 : 
 len 
 ( 
 written_files 
 ), 
 "nextContinuationToken" 
 : 
 cont 
 if 
 pages 
> = 
 MAX_PAGES 
 else 
 None 
 } 
 if 
 __name__ 
 == 
 "__main__" 
 : 
 print 
 ( 
 lambda_handler 
 ()) 
 

  { 
  
 "Version" 
 : 
  
 "2012-10-17" 
 , 
  
 "Statement" 
 : 
  
 [ 
  
 { 
  
 "Effect" 
 : 
  
 "Allow" 
 , 
  
 "Action" 
 : 
  
 [ 
 "s3:GetObject" 
 ], 
  
 "Resource" 
 : 
  
 "arn:aws:s3:::<your-bucket>/*" 
  
 }, 
  
 { 
  
 "Effect" 
 : 
  
 "Allow" 
 , 
  
 "Action" 
 : 
  
 [ 
 "s3:ListBucket" 
 ], 
  
 "Resource" 
 : 
  
 "arn:aws:s3:::<your-bucket>" 
  
 } 
  
 ] 
 }