本文共 13528 字，大约阅读时间需要 45 分钟。
python中flask
在 ，我向您介绍了一个简单应用程序的构建，以使用Threat Stack Webhook并将警报存档到AWS S3。 在本文中，我将深入探讨Python异常处理以及如何以安全的方式进行处理。
我在上一篇文章中编写了尽可能简单易懂的代码，但是如果我的应用程序出现问题怎么办？ 我没有包含任何错误或异常处理。 如果出现问题（例如，说您遇到了错误或收到了错误的数据），您将无法在应用程序中对其进行任何处理。 该应用程序不会返回可解析的JSON（JavaScript对象表示法）响应，而只会向后吐出嵌入HTML文档中的回溯跟踪。 然后，将请求发送到您的服务的实体将试图找出可能出了什么问题。
您需要处理什么？
一些智慧的话：
分布式系统是指您甚至不知道存在的计算机故障都会使您自己的计算机无法使用的系统。
— ，计算机科学家，2013年AM图灵奖得主。
您可以通过在前面的Lamport引用中将“计算机”与“服务”一起开始。 您的应用程序与Threat Stack和AWS S3进行通信。 与其中任何一个通信失败都会导致您自己的服务失败。 服务关闭，响应失败或返回意外响应可能是导致失败的原因。 任何数量的问题都可能导致系统之间的通信失败。
您还需要处理输入验证。 您的服务有两个不同的输入请求：
- 向服务发送警报需要发送和解析JSON文档。
- 搜索警报可以采用可选的日期参数。
通过简单的错误（例如错别字或对所需内容的误解），您对服务的输入可能不是您期望的。 更糟糕的是，有些人会故意发送错误的数据以查看会发生什么。 是一种在应用程序渗透测试中使用的技术，该技术将格式错误或半格式的数据发送到服务以发现错误。
可能发生的最坏情况是什么？
除了是不可靠的，经常中断的服务之外？ 我之前提到过，由于错误，应用程序将返回回溯。 让我们看看将不可解析的日期发送到您的服务时会发生什么：
您正在将自己的代码返回给请求者。 此代码是合理的良性代码，因此请看另一个示例。 如果存在威胁堆栈通信问题：可能完全随机发生（尽管希望不会发生）的问题，则会出现：
您正在泄漏与之交谈的服务的位置，如果开发人员使用了不良做法，您甚至可能会将您的API密钥泄漏给了一个随机的人。
异常捕获和处理
既然您知道了在应用程序中处理异常为什么很重要，那么我将把重点放在如何正确处理它们上。 当您开始处理异常时，您想完成以下任务：
- 找出可能出问题的地方
- 向客户返回有用信息
- 不要泄露太多信息
我承认，直到我写这篇文章并最终进行更正之前，我所做的许多事情都是危险的甚至是错误的。 在寻找答案时，我发现许多其他人对如何正确地做事情也有类似的疑问。 即使您认为这是一个琐碎的话题，为什么不复习呢？
在app.models.threatstack中捕获异常
我将逐步介绍该模块的一部分，以突出一些不同的情况供您处理。 这是用于从警报详细信息的功能：
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中（）进入块，将捕获的异常类型requests.exceptions.RequestException：
try : resp = requests. get ( alerts_url , headers = { 'Authorization' : THREATSTACK_API_KEY } ) except requests. exceptions . RequestException as e: ` Pass 如果失败，则可以执行您认为必要的任何其他操作。 如果使用数据库，则可能会回滚事务。 您可能还希望记录该错误以供以后分析。 （如果您已经为该应用程序编写了日志记录组件，则可能会这样做。）请注意，您正在指定要捕获的异常类型。 不要掩盖所有例外 。 您可能会想这样做以节省时间，但是这可能会使您的生活更加艰难，因为您发现自己无法理解应用程序为什么会失败。 现在花点时间了解您的应用程序为什么会失败以及由于什么原因。
如果应用程序无法与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的文档很好地 。 由于我们应用程序的简单性，您只需要对其稍作调整。 首先将与错误类相关联。 让我们在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 ( ) : #@ s3. errorhandler ( Exception ) def handle_error ( error ) : #
这对于大型应用程序非常有用，大型应用程序可能需要更多的组织和需要自己处理错误的不同视图，但让我们简化代码。 相反，您将添加一个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文档，而不是引发回溯并揭示代码以及可能返回的敏感信息。
最后的想法
您已使从Thrackstack到s3的服务对故障的恢复能力更强，但您可能还会看到我们还有很多事情要做。 在下一篇文章中，我将讨论日志记录。
这篇文章 。
本文最初出现在 。 经许可重新发布。
翻译自:
python中flask
转载地址：http://jzdzd.baihongyu.com/