Python 请求使用特定令牌重定向

Question

import config
import datetime
import io
import json
import os
import pandas as pd
from requests_kerberos import HTTPKerberosAuth, OPTIONAL
import requests
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
import time
import uuid
guid =              str(uuid.uuid4())
engine =            create_engine('postgresql://<user>:<pwd>@igers.db.com:8192/ibdrs?sslmode=require')
path =              'D:/Python-Scripts/SIM/bin/'
jsonFiles =         os.listdir(path)
maxSnapshotQuery =  '''SELECT MAX(Snapshot) AS Snapshot FROM %s''' 
#disable requests SSL warnings
requests.packages.urllib3.disable_warnings()
def createSession() :
global session
sessionURL = 'https://api.com/users/whoami'
session = requests.session()
session.get(sessionURL, auth=HTTPKerberosAuth(mutual_authentication=OPTIONAL), verify=False)
return session
def createJob(session) :
global data
#use the Session to send Json to the Jobs API
simURL = 'https://api.com/jobs'
r = session.post(simURL, json=rawJson, verify=False)
rjson = r.json()
docID = rjson['id']
#Determine Job status
docStatusURL = simURL+'/'+docID
docStatusJson = session.get(docStatusURL, verify=False)
docStatus = docStatusJson.json()['status']

while docStatus != 'Complete' :
time.sleep(10)
docStatusJson = session.get(docStatusURL, verify=False)
docStatus = docStatusJson.json()['status']
if (docStatus == 'Failed') :
break
print('Job Status : %s' %docStatusURL)
exit()
continue
#Once the job is complete, retrieve the attachment ID
attachments = docStatusJson.json()['attachments']
fileID = attachments[0]['id']
#set the file download location and retrieve the file
docURL = 'https://api.com/jobs/%s/attachments/%s' % (docID, fileID)
data = session.get(docURL, verify=False)
return data
def loadData(data, engine) :
stream = data.content
try:
df = pd.read_csv(io.StringIO(stream.decode('utf-8')))
df['Snapshot'] = datetime.datetime.now()
df.where(df.notnull(),None)
dateCol = [col for col in df.columns if 'Date' in col or 'date_' in col or '_date' in col]
for col in dateCol:
df[col] = pd.to_datetime(df[col])
if file.split(".")[0] in ["DCE"] :
df.to_sql(file.split(".")[0], engine, schema = 'sim_and_tt', if_exists='append', index=False)
else :
df.to_sql(file.split(".")[0], engine, schema = 'sim_and_tt', if_exists='replace', index=False)
except SQLAlchemyError as e:
print('Loading data to the database failed for ' + file)
error = str(e.__dict__['orig'])
print(error)

for file in jsonFiles:
tableName = file.split(".")[0]
try:
snapshotDate = pd.read_sql_query(maxSnapshotQuery %tableName, engine, parse_dates=['Snapshot'])
except:
snapshotDate = None
if tableName in ["DCE"] and snapshotDate.Snapshot[0].date() == datetime.datetime.now().date() and snapshotDate.Snapshot[0] < datetime.datetime.now() :
try:
print('attempting to remove the latest snapshot from %s and update...' %tableName)
with engine.connect() as conn:
conn.execute('''DELETE FROM ibdrs."%s" WHERE "Snapshot" = (SELECT MAX("Snapshot") FROM ( SELECT * FROM "%s") AS Get_Snapshot )''' % (tableName, tableName) )
print('Snapshot removed.... preparing for update')
except Exception as err:
print(err)
print(tableName + ' Deleting the snapshot for the same day failed')        
if snapshotDate is None or snapshotDate.Snapshot[0].date() != datetime.datetime.now().date() :
print('loading data for %s' %tableName)
with open(path+file) as f :
try:
rawJson = json.load(f)
rawJson['authorizations'][0]['id'] = guid
except:
print(f.name + ' Check the JSON within')
continue

createSession()
createJob(session)
loadData(data, engine)
print(tableName + ' Load Completed')
else:
print('Data already loaded for today for ' + tableName)

我可以创建一个会话，并在 cookiejar 中获得 3 个不同的令牌。 一个sso_token、sentry_token和一个sso_rfp。 该 API 过去没有 MFA，但他们最近实现了 MFA。 此 MFA 需要重定向，但在代码的这一部分之前，它实际上不会发生：

docURL = 'https://api.com/jobs/%s/attachments/%s' % (docID, fileID)
data = session.get(docURL, verify=False)

我可以通过脚本进行设置，我可以获取文档ID，我可以使用DocID检查文档状态，我可以获取文件ID。 一旦我去实际获取数据，重定向就会发生，它会给出一些 302，然后是 401，直到第 5 次重试最终给出最终的 401。 我正在尝试弄清楚如何为重定向提供sentry_token，因为这显然是它所需要的。 相反，它试图使用sso_token。 如何通过重定向为正确的域使用正确的令牌？

Answer 1

我认为您可以尝试两种方法：

使用 session.cookies.copy(( 将返回一个存储会话 cookie 的 RequestsCookieJar(( 对象。然后，您可以使用 .pop(( 方法从该 cookie jar 中删除 sso 令牌。之后，您可以使用没有 sso 令牌的新 cookie jar 提供cookies关键字 arg。

sso_less_cookies = session.cookies.copy()
sso_less_cookies.pop('sso_token')
session.get(session.get(docURL, verify=False, cookies=sso_less_cookies)

但是，如果您只需要哨兵令牌，则可以尝试创建一个空的cookie罐，并添加原始的哨兵令牌。

cookies_dict = requests.utils.dict_from_cookiejar(session.cookies)
new_cookies = request.session.RequestsCookieJar()
new_cookies.set('sentry_token', cookies_dict['sentry_token'],domain='example.org',path='/cookies')
session.get(docURL, verify=False, cookies=new_cookies)