在我的上一篇 Python Flask 文章中,我带您了解了如何构建一个简单的应用程序,以接收 Threat Stack webhook 并将警报存档到 AWS S3。在这篇文章中,我将深入探讨 Python 异常处理以及如何在安全的方式下进行处理。
我在上一篇文章中编写的代码尽可能简单易懂,但是如果我的应用程序出现问题会怎么样?我没有包含任何错误或异常处理。如果出现问题——例如,假设您遇到错误或收到错误的数据——您无法在应用程序中做任何事情。该应用程序不会返回可解析的 JSON(JavaScript 对象表示法)响应,而是会吐出一个嵌入在 HTML 文档中的回溯。然后,向您的服务发送请求的实体会尝试弄清楚可能出了什么问题。
您需要处理什么?
一些智慧之言
分布式系统是指,即使您不知道存在的计算机出现故障,也可能导致您自己的计算机无法使用。
——Leslie Lamport,计算机科学家和 2013 年 A.M. 图灵奖获得者。
您可以从将 Lamport 引文中的“计算机”替换为“服务”开始。您的应用程序与 Threat Stack 和 AWS S3 通信。与其中任何一个通信失败都可能导致您自己的服务失败。失败可能是由服务宕机、无响应或返回意外响应引起的。任何数量的问题都可能导致系统之间的通信失败。
您还需要处理输入验证。您的服务有两个不同的请求需要输入
- 向服务发送警报需要发送和解析 JSON 文档。
- 搜索警报可以接受可选的日期参数。
您服务的输入可能与您期望的不符,这可能是由于简单的错误,例如拼写错误或对所需内容的误解。更糟糕的是,有些人会故意发送错误的数据以查看会发生什么。 模糊测试 是一种应用程序渗透测试中使用的技术,其中将格式错误或半格式化的数据发送到服务以发现错误。
最坏的情况是什么?
除了成为一个经常崩溃的不可靠服务之外?我之前提到过,如果发生错误,应用程序将返回回溯。让我们看看当向您的服务发送无法解析的日期时会发生什么
您正在将您自己的代码返回给请求者。这段代码相当良性,所以看看另一个例子。如果 Threat Stack 通信出现问题:一个可能完全随机发生的问题(尽管希望不会),这将出现
您正在泄露您正在与之通信的服务的位置,如果开发人员使用了不良实践,您甚至可能已将您的 API 密钥泄露给随机人员。
异常捕获和处理
既然您知道为什么在应用程序中处理异常很重要,那么我将重点介绍如何正确处理它们。当您开始处理异常时,您想要完成以下操作
- 确定可能出错的地方
- 向客户端返回有用的信息
- 不要泄露太多信息
我承认,直到现在,在写这篇文章之前,我做了很多危险甚至不正确的事情,我终于做出了更正。在寻找答案时,我发现许多其他人也有关于如何正确做事的类似问题。即使您认为这是一个微不足道的话题,为什么不复习一下呢?
在 app.models.threatstack 中捕获异常
我将逐步介绍此模块的一部分,以突出显示您需要处理的几种不同情况。这是您从 Threat Stack 获取给定警报 ID 的警报详细信息的函数
def get_alert_by_id(alert_id):
'''
Retrieve an alert from Threat Stack by alert ID.
'''
alerts_url = '{}/alerts/{}'.format(THREATSTACK_BASE_URL, alert_id)
resp = requests.get(
alerts_url,
headers={'Authorization': THREATSTACK_API_KEY}
)
return resp.json()
该函数很简单。它构造一个 URL,向 Threat Stack 发出请求,并返回响应的 JSON 内容。那么可能出什么问题呢?在这三个语句中,有两个很容易出错。在向 Threat Stack 发出请求时,可能会发生导致失败的通信错误。如果您确实收到了响应,您希望解析 JSON 文档。如果响应中没有 JSON 文档怎么办?
让我们从对 Threat Stack 的请求失败开始。将 request.get() 放入一个 try/except 块,该块将捕获异常类型 requests.exceptions.RequestException
try:
resp = requests.get(
alerts_url,
headers={'Authorization': THREATSTACK_API_KEY}
)
except requests.exceptions.RequestException as e:
` Pass
如果失败,这将使您可以执行您认为必要的任何其他操作。如果您正在使用数据库,您可能需要回滚事务。您可能还想记录错误以供以后分析。(如果您已经为此应用程序编写了日志记录组件,您可能会这样做。)请注意,您正在指定要捕获的异常类型。不要 blanket 捕获所有异常。您可能会为了节省时间而这样做,但这可能会使您以后的生活更加艰难,因为您会发现自己无法理解为什么您的应用程序会失败。现在花时间了解您的应用程序可能因何种原因而失败。
如果应用程序无法与 Threat Stack 通信,您想做什么?您将引发一个新的异常。这称为捕获并重新引发。此技术使组织异常处理变得更容易一些。您将在 app.models.threatstack 模块内定义一组异常类,以描述可能出错的情况。这样做将在稍后添加处理程序到应用程序并告诉它如何处理来自 app.models.threatstack 模块的异常时更容易。
您将从添加两个异常类开始。第一个是基类异常,它继承了基本 Python Exception 类。每个后续的异常类都将继承新的基类异常。起初,这可能看起来只是额外的工作,但从长远来看,它将很有用。下一个类将用于请求失败。您甚至会添加一个 Threat Stack API 错误,您稍后会用到它。您希望类名称具有描述性,以便您仅通过阅读它就能理解应用程序失败的原因
class ThreatStackError(Exception):
'''Base Threat Stack error.'''
class ThreatStackRequestError(ThreatStackError):
'''Threat Stack request error.'''
class ThreatStackAPIError(ThreatStackError):
'''Threat API Stack error.'''
有了 Exception 类,您可以捕获并重新引发异常
try:
resp = requests.get(
alerts_url,
headers={'Authorization': THREATSTACK_API_KEY}
)
except requests.exceptions.RequestException as e:
exc_info = sys.exc_info()
raise ThreatStackRequestError, ThreatStackRequestError(e), exc_info[2]
捕获异常后会发生什么?为什么您不这样做呢?
except requests.exceptions.RequestException as e:
raise ThreatStackRequestError(e.args)
当人们捕获并重新引发异常时,这个错误非常常见。如果您执行了上述操作,您将丢失应用程序回溯。检查回溯会显示您进入了 get_alert_by_id(),然后您引发了错误。您看不到 request.get() 失败的更深层次的上下文。前面的示例是在 Python 2 中捕获并重新引发错误的正确方法。 您的代码将抛出一个以您知道的类命名的异常,它将为您提供导致异常的代码跟踪,以便您可以更好地调试它。
您已发出请求,与 Threat Stack 正确通信,并准备好在此函数末尾返回响应
return resp.json()
这里可能会出什么问题?首先,响应可能不是 JSON 正文,这将导致您在尝试解析它时抛出异常。API 始终应该返回 JSON,即使在错误时也是如此,但仍然有可能发生意外错误。也许应用程序问题会在错误时像您的应用程序现在所做的那样喷出回溯。也许负载均衡器有问题并返回带有“服务不可用”页面的 503 错误。API 故障也可能发生。您可能被发回了一个 JSON 响应,该响应完全可以解析,只是告诉您您的请求因某种原因而失败。例如,当您尝试检索不存在的警报时。简而言之,您需要确保您的请求返回了成功的响应。如果您没有获得成功的响应,您将引发错误。您可能会收到通信错误或 API 错误,因此根据您收到的内容,您将引发 ThreatStackRequestError 或 ThreatStackAPIError
if not resp.ok:
if 'application/json' in resp.headers.get('Content-Type'):
raise ThreatStackAPIError(resp.reason,
resp.status_code,
resp.json()
)
else:
raise ThreatStackRequestError(resp.reason, resp.status_code)
return resp.json()
如果请求成功,resp.ok 将为 True。如果不是,那么您将尝试确定发生了哪种类型的故障:通信故障还是 API 故障?您将使用一种非常简单的方法来区分差异。如果响应标头指示 JSON,则假设您能够与 API 通信并且 API 向您发送了错误。否则,假设沿途发生了其他错误,您从未到达 Threat Stack API,并且它是通信错误。
处理异常
到目前为止,您一直在捕获异常只是为了重新引发新的异常。您可能会觉得您离开始的地方并没有远多少。您只是在引发异常并将回溯返回给客户端,但使用的是您自己的类名。
您仍然在泄露代码,可能泄露机密,并向某人提供比您真正想要的更多的关于您的环境的情报。现在您需要开始处理这些异常。
Flask 的文档提供了 处理异常的良好概述。由于我们应用程序的简单性,您只需稍微调整一下即可。首先将 HTTP 状态代码 与您的错误类关联起来。让我们回顾一下 app.models.threatstack 中的 Threat Stack 错误类
class ThreatStackError(Exception):
'''Base Threat Stack error.'''
class ThreatStackRequestError(ThreatStackError):
'''Threat Stack request error.'''
class ThreatStackAPIError(ThreatStackError):
'''Threat API Stack error.'''
当您的服务尝试与 Threat Stack 通信并且发生意外情况时,您会引发这些异常。这些可以被认为是 500 级别的服务器错误。(注意:您可以提出一个论点,即传递给 get_alert_by_id() 的无效警报 ID 会引发 ThreatStackAPIError 异常,实际上应该是 400 Bad Request,但我不太关心。我自己的偏好是简单地将模型级别的异常视为 500 级别,并将视图级别的异常视为 400 级别。)还记得我建议创建基类 ThreatStackError 吗?这是您第一次使用它
class ThreatStackError(Exception):
'''Base Threat Stack error.'''
status_code = 500
class ThreatStackRequestError(ThreatStackError):
'''Threat Stack request error.'''
class ThreatStackAPIError(ThreatStackError):
'''Threat API Stack error.'''
对在 app.models.s3 和 app.views.s3 中添加 status_codes 重复此过程。
现在您的错误类具有 HTTP 状态代码,您将为应用程序异常添加处理程序。Flask 的文档使用 errorhandler() 装饰器。您可以将装饰器和一个函数添加到 app.view.s3 模块,就像您向应用程序添加另一个端点一样
@s3.route('/status', methods=['GET'])
def is_available():
# <SNIP>
@s3.errorhandler(Exception)
def handle_error(error):
# <SNIP>
这对于较大的应用程序来说非常棒,这些应用程序可能需要更多的组织和不同的视图,这些视图需要它们自己的错误处理,但让我们保持您的代码更简单一些。相反,您将添加一个用于处理错误的单个 Flask 蓝图,它将处理所有应用程序异常
'''Application error handlers.'''
from flask import Blueprint, jsonify
errors = Blueprint('errors', __name__)
@errors.app_errorhandler(Exception)
def handle_error(error):
message = [str(x) for x in error.args]
status_code = error.status_code
success = False
response = {
'success': success,
'error': {
'type': error.__class__.__name__,
'message': message
}
}
return jsonify(response), status_code
这是一个好的开始,但您将进行额外的调整。我们假设所有 Exception 对象都有一个 status_code 属性,但这根本不是真的。我们希望认为我们已准备好捕获代码中每个可能的异常情况,但人会犯错。出于这个原因,您将有两个错误处理函数。一个将处理您知道的错误类(这是我们的基类异常),另一个将用于意外错误。
另一个需要注意的重要事项是,应用程序盲目地返回与您捕获的错误关联的消息。您仍然有泄露有关您的基础设施、您的应用程序如何工作或您的机密的风险。在这个特定的应用程序案例中,您不必太担心,因为您知道您捕获和重新引发的异常类型以及这些异常返回的信息。对于那些您没有预料到的异常,您始终返回相同的错误消息作为预防措施。我将在以后的文章中重新讨论这一点,届时我将讨论日志记录。因为此应用程序当前没有日志记录,所以您依赖错误响应来提供高度描述性。
当您返回 API 错误时,问问自己谁将使用您的服务。请求者是否需要知道您返回的那么多信息?开发人员可能会感谢添加的上下文来帮助他们调试自己的服务。外部第三方可能不需要知道您的后端是如何失败的。
'''Application error handlers.'''
from app.models.s3 import S3ClientError
from app.models.threatstack import ThreatStackError
from flask import Blueprint, jsonify
errors = Blueprint('errors', __name__)
@errors.app_errorhandler(S3ClientError)
@errors.app_errorhandler(ThreatStackError)
def handle_error(error):
message = [str(x) for x in error.args]
status_code = 500
success = False
response = {
'success': success,
'error': {
'type': error.__class__.__name__,
'message': message
}
}
return jsonify(response), status_code
@errors.app_errorhandler(Exception)
def handle_unexpected_error(error):
status_code = 500
success = False
response = {
'success': success,
'error': {
'type': 'UnexpectedException',
'message': 'An unexpected error has occurred.'
}
}
return jsonify(response), status_code
最后,您将在 app 模块中将此蓝图连接到应用程序。您添加了一个名为 _initialize_errorhandler() 的附加函数,它将导入蓝图并将其添加到您的应用程序
def _initialize_errorhandlers(application):
'''
Initialize error handlers
'''
from app.errors import errors
application.register_blueprint(errors)
def create_app():
'''
Create an app by initializing components.
'''
application = Flask(__name__)
_initialize_errorhandlers(application)
_initialize_blueprints(application)
# Do it!
return application
现在,当应用程序抛出异常时,您有了功能性的错误处理,因此该应用程序不会抛出回溯并泄露代码以及可能返回敏感信息,而是返回一个描述错误的 JSON 文档。
最终想法
您已经使您的 threatstack-to-s3 服务对故障的抵抗力大大提高,但您可能也看到我们需要做更多的事情。在即将发布的文章中,我将讨论日志记录。
本文最初出现在 Threat Stack 博客上。经许可转载。
1 条评论