Requests 重定向与历史

import requests

# 默认情况下，Requests会自动跟随重定向
response = requests.get('http://httpbin.org/redirect/2')

# 即使原始请求返回302，最终响应是200
print(f"最终状态码: {response.status_code}")  # 200
print(f"请求URL: {response.url}")           # 最终URL
print(f"历史记录: {response.history}")      # 重定向历史

# 查看重定向链中的每个响应
for resp in response.history:
    print(f"重定向 {resp.status_code}: {resp.url}")

import requests

def analyze_redirects(url):
    """分析URL的重定向链"""

    response = requests.get(url, allow_redirects=True)

    print(f"分析: {url}")
    print(f"最终URL: {response.url}")
    print(f"最终状态码: {response.status_code}")
    print(f"重定向次数: {len(response.history)}")
    print()

    if response.history:
        print("重定向历史:")
        print("-" * 80)

        for i, resp in enumerate(response.history, 1):
            print(f"{i}. 状态码: {resp.status_code}")
            print(f"   请求URL: {resp.url}")
            print(f"   重定向到: {resp.headers.get('Location', '未知')}")
            print(f"   响应头: {dict(resp.headers)}")
            print("-" * 80)

    return response

# 使用示例
if __name__ == "__main__":
    # 测试重定向链
    result = analyze_redirects('http://httpbin.org/redirect/3')

    # 获取最终响应的详细信息
    print(f"\n最终响应:")
    print(f"URL: {result.url}")
    print(f"内容类型: {result.headers.get('content-type')}")
    print(f"内容长度: {len(result.content)} 字节")

import requests

# 发送请求并跟随重定向
response = requests.get('https://httpbin.org/redirect/2')

# 检查是否有重定向历史
if response.history:
    print("发现重定向历史:")

    # 遍历所有历史响应
    for i, historical_response in enumerate(response.history, 1):
        print(f"\n第 {i} 次重定向:")
        print(f"  状态码: {historical_response.status_code}")
        print(f"  请求URL: {historical_response.url}")
        print(f"  请求方法: {historical_response.request.method}")
        print(f"  响应头: {dict(historical_response.headers)}")
        print(f"  重定向目标: {historical_response.headers.get('Location')}")
        print(f"  请求耗时: {historical_response.elapsed}")

        # 也可以访问请求对象
        request = historical_response.request
        print(f"  原始请求头: {dict(request.headers)}")
        print(f"  原始请求体: {request.body}")
else:
    print("没有重定向历史")

print(f"\n最终响应:")
print(f"URL: {response.url}")
print(f"状态码: {response.status_code}")

import requests

# 禁用自动重定向
response = requests.get('http://httpbin.org/redirect/2', allow_redirects=False)

print(f"状态码: {response.status_code}")  # 302 (不是200)
print(f"响应URL: {response.url}")       # 原始URL，不是最终URL
print(f"Location头: {response.headers.get('Location')}")
print(f"历史记录: {response.history}")  # 空列表

# 手动处理重定向
if response.status_code in [301, 302, 303, 307, 308]:
    redirect_url = response.headers['Location']
    print(f"需要重定向到: {redirect_url}")

    # 手动发送新的请求
    new_response = requests.get(redirect_url)
    print(f"最终状态码: {new_response.status_code}")
    print(f"最终URL: {new_response.url}")

import requests
from requests.exceptions import TooManyRedirects

# 设置最大重定向次数
try:
    # 尝试访问一个有很多重定向的URL
    response = requests.get('http://httpbin.org/redirect/5', max_redirects=3)
except TooManyRedirects as e:
    print(f"重定向次数过多: {e}")
    print(f"允许的最大重定向次数: 3")

    # 可以获取到最后一个响应
    # 注意：在异常处理中，可以通过e.response访问最后一个响应
    if hasattr(e, 'response') and e.response:
        print(f"最后一个响应的状态码: {e.response.status_code}")
        print(f"最后一个响应的URL: {e.response.url}")
        print(f"已经发生的重定向次数: {len(e.response.history)}")

# 或者通过Session设置默认的最大重定向次数
session = requests.Session()
session.max_redirects = 5  # 设置Session级别的最大重定向次数

try:
    response = session.get('http://httpbin.org/redirect/10')
except TooManyRedirects:
    print("Session级别的重定向限制生效")

import requests

def redirect_hook(response, **kwargs):
    """重定向钩子函数"""
    print(f"发生重定向:")
    print(f"  从: {response.url}")
    print(f"  状态码: {response.status_code}")
    print(f"  重定向到: {response.headers.get('Location')}")
    print(f"  重定向历史长度: {len(response.history)}")

    # 可以在这里添加自定义逻辑
    # 例如：记录日志、修改请求、检查重定向目标等

    return response

# 创建自定义钩子
hooks = {'response': [redirect_hook]}

# 发送请求并监控重定向
response = requests.get(
    'http://httpbin.org/redirect/3',
    hooks=hooks,
    allow_redirects=True
)

print(f"\n最终结果:")
print(f"最终URL: {response.url}")
print(f"最终状态码: {response.status_code}")
print(f"总重定向次数: {len(response.history)}")

# 也可以在Session级别设置钩子
session = requests.Session()
session.hooks['response'].append(redirect_hook)

# Session的所有请求都会调用钩子
session.get('http://httpbin.org/redirect/2')

import requests
from urllib.parse import urljoin, urlparse

class ManualRedirectHandler:
    """手动重定向处理器"""

    def __init__(self, max_redirects=10, timeout=30):
        self.max_redirects = max_redirects
        self.timeout = timeout
        self.session = requests.Session()
        self.redirect_history = []

    def resolve_url(self, url, original_method='GET', data=None, headers=None):
        """手动解析URL重定向链"""

        current_url = url
        current_method = original_method
        current_data = data
        redirect_count = 0

        while redirect_count < self.max_redirects:
            # 发送请求（不自动重定向）
            response = self.session.request(
                method=current_method,
                url=current_url,
                data=current_data,
                headers=headers,
                allow_redirects=False,
                timeout=self.timeout
            )

            # 记录历史
            self.redirect_history.append({
                'url': current_url,
                'method': current_method,
                'status_code': response.status_code,
                'location': response.headers.get('Location')
            })

            # 检查是否是重定向
            if response.status_code in [301, 302, 303, 307, 308]:
                redirect_count += 1

                # 获取重定向目标
                location = response.headers.get('Location')
                if not location:
                    raise ValueError(f"重定向响应缺少Location头: {response.status_code}")

                # 解析相对URL
                if not urlparse(location).netloc:
                    location = urljoin(current_url, location)

                # 根据重定向类型调整HTTP方法
                if response.status_code in [301, 302, 303]:
                    # 对于这些状态码，POST应该变为GET
                    if current_method == 'POST':
                        current_method = 'GET'
                        current_data = None  # GET请求没有请求体
                # 307和308保持HTTP方法不变

                print(f"重定向 {redirect_count}: {response.status_code} {current_url} → {location}")
                current_url = location

            else:
                # 不是重定向，返回最终响应
                print(f"解析完成，共 {redirect_count} 次重定向")
                return response

        # 超过最大重定向次数
        raise requests.exceptions.TooManyRedirects(
            f"超过最大重定向次数: {self.max_redirects}"
        )

    def get_redirect_chain(self):
        """获取重定向链"""
        return self.redirect_history

# 使用示例
if __name__ == "__main__":
    handler = ManualRedirectHandler(max_redirects=5)

    try:
        # 手动解析重定向
        response = handler.resolve_url(
            'http://httpbin.org/redirect/3',
            original_method='GET'
        )

        print(f"\n最终响应:")
        print(f"URL: {response.url}")
        print(f"状态码: {response.status_code}")

        print(f"\n重定向链:")
        for i, entry in enumerate(handler.get_redirect_chain(), 1):
            print(f"{i}. {entry['method']} {entry['url']} → {entry['status_code']} → {entry['location']}")

    except requests.exceptions.TooManyRedirects as e:
        print(f"重定向次数过多: {e}")
    except Exception as e:
        print(f"错误: {e}")

import requests
from urllib.parse import urlparse, urljoin
from typing import List

class SecureRedirectHandler:
    """安全重定向处理器"""

    def __init__(self, allowed_domains: List[str] = None,
                 allow_cross_protocol: bool = False,
                 max_redirects: int = 10):

        self.allowed_domains = allowed_domains or []
        self.allow_cross_protocol = allow_cross_protocol
        self.max_redirects = max_redirects
        self.session = requests.Session()

    def is_safe_redirect(self, current_url: str, redirect_url: str) -> bool:
        """检查重定向是否安全"""

        current = urlparse(current_url)
        redirect = urlparse(redirect_url)

        # 1. 检查协议降级 (HTTPS → HTTP)
        if current.scheme == 'https' and redirect.scheme == 'http':
            if not self.allow_cross_protocol:
                print(f"安全警告: 协议降级 {current.scheme} → {redirect.scheme}")
                return False

        # 2. 检查域名白名单
        if self.allowed_domains:
            if redirect.netloc not in self.allowed_domains:
                print(f"安全警告: 重定向到未允许的域名 {redirect.netloc}")
                return False

        # 3. 检查相对URL解析
        if not redirect.netloc and redirect.path.startswith('//'):
            print(f"安全警告: 可能的协议相对URL {redirect_url}")
            return False

        return True

    def safe_get(self, url: str, **kwargs) -> requests.Response:
        """安全地发送GET请求并处理重定向"""

        current_url = url
        redirect_count = 0

        while redirect_count < self.max_redirects:
            # 发送请求（不自动重定向）
            response = self.session.get(
                current_url,
                allow_redirects=False,
                **kwargs
            )

            # 检查是否是重定向
            if response.status_code in [301, 302, 303, 307, 308]:
                redirect_count += 1

                # 获取重定向目标
                location = response.headers.get('Location')
                if not location:
                    raise ValueError("重定向响应缺少Location头")

                # 解析URL
                if not urlparse(location).netloc:
                    location = urljoin(current_url, location)

                # 安全检查
                if not self.is_safe_redirect(current_url, location):
                    raise ValueError(f"不安全的重定向: {current_url} → {location}")

                print(f"安全重定向 {redirect_count}: {location}")
                current_url = location

            else:
                # 不是重定向，返回最终响应
                return response

        raise requests.exceptions.TooManyRedirects(
            f"超过最大重定向次数: {self.max_redirects}"
        )

# 使用示例
if __name__ == "__main__":
    # 创建安全重定向处理器
    handler = SecureRedirectHandler(
        allowed_domains=['httpbin.org', 'example.com'],
        allow_cross_protocol=False,
        max_redirects=5
    )

    try:
        response = handler.safe_get('https://httpbin.org/redirect/2')
        print(f"安全获取成功: {response.status_code}")
    except ValueError as e:
        print(f"安全错误: {e}")
    except requests.exceptions.TooManyRedirects as e:
        print(f"重定向次数过多: {e}")

import requests
from requests.exceptions import TooManyRedirects, RequestException
from urllib.parse import urlparse

class BestPracticeRedirectHandler:
    """遵循最佳实践的重定向处理器"""

    @staticmethod
    def safe_request(url, method='GET', data=None, **kwargs):
        """
        安全的请求函数，包含重定向最佳实践

        参数:
            url: 请求URL
            method: HTTP方法
            data: 请求数据
            **kwargs: 其他requests参数

        返回:
            Response对象或None（如果失败）
        """

        # 最佳实践配置
        config = {
            'allow_redirects': True,      # 启用自动重定向
            'max_redirects': 10,          # 合理的重定向限制
            'timeout': (3.05, 30),        # 连接和读取超时
            'verify': True,               # 启用SSL验证
        }

        # 合并用户配置
        config.update(kwargs)

        try:
            # 发送请求
            response = requests.request(
                method=method,
                url=url,
                data=data,
                **config
            )

            # 检查重定向历史
            if response.history:
                # 记录重定向信息
                print(f"请求经过了 {len(response.history)} 次重定向")

                # 检查是否有不安全的协议降级
                original_url = urlparse(url)
                final_url = urlparse(response.url)

                if original_url.scheme == 'https' and final_url.scheme == 'http':
                    print("警告: 发生了HTTPS到HTTP的协议降级")

            # 验证最终URL
            final_domain = urlparse(response.url).netloc
            expected_domains = ['example.com', 'api.example.com']  # 你的允许域名

            if final_domain not in expected_domains:
                print(f"警告: 最终域名 {final_domain} 不在允许列表中")

            return response

        except TooManyRedirects as e:
            print(f"错误: 重定向次数过多 ({config['max_redirects']} 次)")

            # 可以获取最后一个响应进行分析
            if hasattr(e, 'response') and e.response:
                print(f"最后一个响应URL: {e.response.url}")
                print(f"重定向历史: {len(e.response.history)} 次")

            return None

        except RequestException as e:
            print(f"请求错误: {e}")
            return None

    @staticmethod
    def analyze_redirect_chain(url):
        """分析URL的重定向链并返回报告"""

        try:
            response = requests.get(url, allow_redirects=True)

            report = {
                'original_url': url,
                'final_url': response.url,
                'final_status_code': response.status_code,
                'redirect_count': len(response.history),
                'redirect_chain': [],
                'security_issues': []
            }

            # 分析重定向链
            previous_url = url
            for i, resp in enumerate(response.history, 1):
                redirect_info = {
                    'step': i,
                    'from': previous_url,
                    'to': resp.headers.get('Location', '未知'),
                    'status_code': resp.status_code,
                    'method_change': None
                }

                # 检查方法变化
                if i > 1:
                    prev_resp = response.history[i-2]
                    if prev_resp.request.method != resp.request.method:
                        redirect_info['method_change'] = {
                            'from': prev_resp.request.method,
                            'to': resp.request.method
                        }

                report['redirect_chain'].append(redirect_info)
                previous_url = resp.headers.get('Location', '未知')

            # 检查安全问题
            for i in range(len(report['redirect_chain'])):
                current = urlparse(report['redirect_chain'][i]['from'])
                next_url = report['redirect_chain'][i]['to']

                if next_url != '未知':
                    nxt = urlparse(next_url)

                    # 检查协议降级
                    if current.scheme == 'https' and nxt.scheme == 'http':
                        report['security_issues'].append(
                            f"步骤 {i+1}: 协议降级 HTTPS → HTTP"
                        )

                    # 检查域名变化
                    if current.netloc != nxt.netloc:
                        report['security_issues'].append(
                            f"步骤 {i+1}: 域名变化 {current.netloc} → {nxt.netloc}"
                        )

            return report

        except Exception as e:
            return {'error': str(e)}

# 使用示例
if __name__ == "__main__":
    # 使用最佳实践函数
    response = BestPracticeRedirectHandler.safe_request(
        'https://httpbin.org/redirect/2',
        method='GET'
    )

    if response:
        print(f"请求成功: {response.status_code}")
        print(f"最终URL: {response.url}")

    # 分析重定向链
    print("\n分析重定向链:")
    report = BestPracticeRedirectHandler.analyze_redirect_chain(
        'https://httpbin.org/redirect/3'
    )

    if 'error' not in report:
        print(f"原始URL: {report['original_url']}")
        print(f"最终URL: {report['final_url']}")
        print(f"重定向次数: {report['redirect_count']}")

        if report['security_issues']:
            print("\n安全问题:")
            for issue in report['security_issues']:
                print(f"  - {issue}")