Over a million developers have joined DZone.
{{announcement.body}}
{{announcement.title}}

Python Flask Exception Handling in a Secure Manner

DZone's Guide to

Python Flask Exception Handling in a Secure Manner

Exception handling is crucial in any language or framework. In this post, we take a look at how to securely handle exceptions in Python's Flask framework.

· Web Dev Zone
Free Resource

Get deep insight into Node.js applications with real-time metrics, CPU profiling, and heap snapshots with N|Solid from NodeSource. Learn more.

Image title

In our last Python Flask blog post, we walked through building a simple application to take in a Threat Stack webhook and archive the alert to AWS S3. In this post, we’ll dive into Python exception handling and how to do it in a secure manner.

The code in the last post was written to be as simple and readable as possible. However, what happens if something goes wrong in our application? There’s no error or exception handling. If something goes wrong — for example, we hit a bug or receive bad data — there’s nothing we can do about it in the application. Instead of returning a parseable JSON response, the app will just spit a backtrace embedded in an HTML document back. The entity sending the request to our service is then left trying to figure out what may have gone wrong.

What Do We Need to Handle?

Image title

We can start by placing “computer” with “service” in the preceding Leslie Lamport quotation. Our application talks to Threat Stack and AWS S3. A failure communicating with either of those can cause our own service to fail. A failure might be caused by the service being down, being unresponsive, or returning an unexpected response. Any number of issues can cause a communication failure between systems.

We also need to handle input validation. Our service has two different requests that take input:

  • Sending alerts to the service requires a JSON document to be sent and parsed.
  • Searching for alerts can take optional date parameters.

The input to our service might not be what we expect through a simple mistake such as a typo or misunderstanding of what’s required. Worse, some people will intentionally send bad data to see what happens. Fuzzing is a technique used in application penetration testing where malformed or semi-formed data is sent to a service to uncover bugs.

What Is the Worst That Could Happen?

Other than being an unreliable service that’s regularly breaking? We mentioned before that on an error the application will return a backtrace. Let’s see what happens when an unparseable date is sent to our service:

Image title

We are returning our own code back to the requester. This code is reasonably benign, so let’s look at another. Here’s what would appear if there was a Threat Stack communication issue: an issue that might (though hopefully not) happen completely at random:

Image title

We are leaking the location of the service we’re talking to. And if a developer had used poor practices, we might have even leaked our API key to a random person.

Exception Catching and Handling

Now that we know why it’s important to handle exceptions in our application, let’s get down to handling them properly. We want to accomplish the following when we start handling exceptions:

  • Identify what could go wrong.
  • Return useful information to the client.
  • Don’t leak too much information.

I’ll admit that until now I did many things dangerously or even incorrectly that in writing this post I finally corrected. While searching for answers, I found that many other people had similar questions about how to do things correctly. Even if you think this is a trivial topic, why not take a refresher?

Catching Exceptions in app.models.threatstack

We will walk through a part of this module because it will highlight a few different situations for us to handle. This is our function for getting 'alert' detail from Threat Stack for a given alert ID.

See the full code at https://github.com/threatstack/threatstack-to-s3/blob/37f11443583b6b0a5be1850d39383495626f7fa3/app/models/threatstack.py; or see the snippet below: 

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()


The function is straightforward. It constructs a URL, makes a request to Threat Stack, and returns the response’s JSON content. So what can wrong? Of those three statements, two can easily go wrong. When making a request to Threat Stack, there may be a communication error that results in failure. If we do get a response, we expect to parse a JSON document. What if there is no JSON document in the response?

Let’s start with a failed request to Threat Stack. We’ll put request.get() into a try/except block that will catch the exception type requests.exceptions.RequestException:

 try:
        resp = requests.get(
            alerts_url,
            headers={'Authorization': THREATSTACK_API_KEY}
        )

    except requests.exceptions.RequestException as e:
        Pass


If we fail, this lets us perform any additional actions that we feel are necessary. If we were working with a database, we might roll back a transaction. We might want to log the error as well for analysis later (we would probably do that if we had already written the logging component for this application). Notice that we specify the exception type to catch. Do not blanket catch all exceptions. You may be tempted to do this to save time, but it will potentially make your life harder down the road as you find yourself unable to understand why your application is failing. Take the time now to understand why your application might fail and for what reasons.

So what do we want to do if the app fails to communicate with Threat Stack? We’re going to raise a new exception. What? This is called catch and reraise. It’s a technique to make organizing exception handling a little easier. We are going to define a set of exception classes inside the app.models.threatstack module that describes what could go wrong. Doing this will make it easier later when we will add a handler to the application and tell it how to handle exceptions from the app.models.threatstack module.

We will start by adding two exception classes. The first is the base exception class which inherits the base Python Exception class. Every subsequent exception class will inherit the new base exception class. At first, it may just seem like extra work, but it will be useful down the road. The next class will be for request failures. We will even add a Threat Stack API error that we will use later on. We want the class name to be descriptive so that just by reading it, we have a clue as to why our application failed:

class ThreatStackError(Exception):
    '''Base Threat Stack error.'''

class ThreatStackRequestError(ThreatStackError):
    '''Threat Stack request error.'''

class ThreatStackAPIError(ThreatStackError):
    '''Threat API Stack error.'''


With the Exception classes in place, let’s now catch and reraise an 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]


What’s going on after we catch the exception? Why didn’t we just do this?

   except requests.exceptions.RequestException as e:
        raise ThreatStackRequestError(e.args)


This is a very common mistake when people catch and reraise exceptions. If you did the above, you lose the application backtrace. Inspecting the backtrace would show that you entered get_alert_by_id() and then you raised an error. We would not see the further context of why request.get() failed. The previous example is the correct way to catch and reraise errors in Python 2. Our code will throw an exception named for a class that we know, and it will give us the code trace that leads to the exception so we can better debug it.

We have made a request, communicated with Threat Stack correctly, and are ready to return the response at the end of this function:

   except requests.exceptions.RequestException as e:
        raise ThreatStackRequestError(e.args)


What can go wrong here? For one thing, the response may not have been a JSON body, which would cause us to throw an exception while attempting to parse it. The API is always supposed to return JSON, even on an error, but it is possible that something might still go unexpectedly wrong. Maybe an application issue spews a backtrace on an error just as our application does right now. Maybe a load balancer has an issue and returns a 503 with a “Service Unavailable” page. API failures can also occur. We might have been sent back a JSON response that’s perfectly parseable only to tell us that our request failed for some reason. For example, we are trying to retrieve an alert that does not exist. Simply put, we need to make sure that our request returned a successful response. If we didn’t get a successful response, we raise an error. We might be given a communication error or an API error, so depending on what we received, we will raise either ThreatStackRequestError or 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()


If the request was successful, resp.ok will be True. If it is not, then we will try to determine what sort of failure occurred: communication or API? We will use a very simple approach to figuring out the difference. If the response header indicates JSON, we will assume we were able to talk to the API and the API sent us an error. Otherwise, we will assume that something else along the way failed and we never made it to the Threat Stack API and that it is a communication error.

Handling Exceptions

So far we have been catching exceptions only to reraise a new exception. It might feel that we are not that much further from where we started. We are just raising exceptions and returning a backtrace to the client, but with our own class name.

Image title

We are still leaking code, potentially leaking secrets, and providing someone with greater intelligence about our environment than we really want to. Now we need to start handling these exceptions.

Flask’s documentation provides a good overview of handling exceptions. We are just going to tweak it slightly due to the simplicity of our application. Let’s start by associating HTTP status codes with our error classes. Let’s revisit our Threat Stack error classes in app.models.threatstack.

See the full code at app.models.threatstack; or see the snippet below: 

class ThreatStackError(Exception):
    '''Base Threat Stack error.'''

class ThreatStackRequestError(ThreatStackError):
    '''Threat Stack request error.'''

class ThreatStackAPIError(ThreatStackError):
    '''Threat API Stack error.'''


We raise these exceptions when our service attempts to talk to Threat Stack and something unexpected happens. These can arguably be considered 500 level server errors (Note: You can make a case that an invalid alert ID passed to get_alert_by_id() which raises a ThreatStackAPIError exception should actually be a 400 Bad Request, but we are not that concerned. My own preference is to simply consider model level exceptions as 500 level and view level exceptions as 400 level). Recall when I suggested creating a base ThreatStackError class? Here’s where we first use it.

See full code at app.models.threatstack, or use the snippet below:

class ThreatStackError(Exception):
    '''Base Threat Stack error.'''
    status_code = 500

class ThreatStackRequestError(ThreatStackError):
    '''Threat Stack request error.'''

class ThreatStackAPIError(ThreatStackError):
    '''Threat API Stack error.'''


Repeat this process for adding status_codes in app.models.s3 and app.views.s3 too.

Now that our error classes have an HTTP status code, let’s add a handler for application exceptions. Flask’s documentation uses the errorhandler() decorator. We would add the decorator and a function to the app.view.s3 module just as if we were adding another endpoint to our application:

See the full code at app.view.s3 or use the snippet below: 

@s3.route('/status', methods=['GET'])
def is_available():


@s3.errorhandler(Exception)
def handle_error(error):


This is great for larger apps which perhaps require more organization and different views that require their own error handling, but we will elect to keep our code a little simpler. Instead, we will add a single Flask blueprint for handling errors which will handle all application exceptions:

See the full code at app.errors or use the following:

'''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


This is good to start with, but let’s make an additional tweak. We are assuming that all Exception objects have a status_code attribute, which is simply not true. We would like to think that we are prepared to catch every possible exception case in our code, but people make mistakes. For that reason, we will have two error handler functions. One will handle the error classes we know about (there's our base exception classes again), and the other will be for unexpected errors.

Another important thing to notice is that the application blindly returns the message associated with errors we catch. We’re still at risk of potentially revealing information about our infrastructure, how our application works, or secrets. In this particular application’s case, we are not as worried because we are aware of the types of exceptions we catch and reraise along with the information those exceptions return. For those exceptions we didn’t anticipate, we always return the same error message as a precaution. We will revisit this in a later post when we discuss logging. Since this application currently has no logging, we are relying on the error response to be highly descriptive.

When you are returning API errors, ask yourself who will be using your service. Does the requester need to know as much as we are returning? A developer might appreciate the added context to help them debug their own service. An external third party probably doesn’t need to know how your backend failed.

See the full code at app.errors; or see the snippet below: 

'''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


Finally, let’s hook this blueprint up to the application in the app module. We add an additional function called _initialize_errorhandler() which will import the blueprint and add it to our application:

See the full code at app, or see the snippet below: 

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)


    return application 


So now we have functional error handling when the application throws an exception, so instead of throwing a backtrace and revealing code as well as potentially returning sensitive information, the app returns a JSON doc that describes the error.

Conclusion

We have made our threatstack-to-s3 service far more resilient to failure, but we have also touched on there being more for us to do. In an upcoming post, we will discuss logging.

If you’d like to see the finished product from this post, take a look here:

Node.js application metrics sent directly to any statsd-compliant system. Get N|Solid

Topics:
python ,flask ,exception handling ,web dev ,tutorial

Published at DZone with permission of Tom McLaughlin, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}