I have a project that has the following structure
app.py
|
endpoints(directory)
__init__.py
endpoints_1.py
endpoints_2.py
auth_endppints.py
auth_functionality_module.py
etc ...
app.py has the following content (simplified)
# Import modules
..... imports
# Set document root
ROOT = os.path.dirname(os.path.abspath(__file__))
path = Path(ROOT)
parent = path.parent.absolute()
# Change settings based on config
config = ConfigParser()
config_auth = ConfigParser()
config.read(vv_settings.CONFIG_DIR)
config_auth.read(str(parent) + "/configuration/auth.ini")
# Get Secret Key
SECRET_KEY = config_auth['secret']['secret_key']
# Database config
SQLALCHEMY_DATABASE_URI = '<URI built in the code>'
"""
Create a parser object locally
"""
parser = request_parser.parser
# initiate
application = Flask(__name__)
api.init_app(application)
# configure application
# By default, show all endpoints (collapsed)
application.config.SWAGGER_UI_DOC_EXPANSION = 'list'
application.config.from_object(__name__)
application.config['SECRET_KEY'] = SECRET_KEY
application.config['PROPAGATE_EXCEPTIONS'] = False
application.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI
application.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
application.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(application)
class User(db.Model):
# Database model, currently very simple
__tablename__ = 'authUsers'
id = db.Column(db.Integer, primary_key=True)
domain = db.Column(db.String(50), index=True)
password_hash = db.Column(db.String(1000))
email = db.Column(db.String(50))
added = db.Column(db.Date)
expires = db.Column(db.Date)
termination = db.Column(db.Integer)
def hash_password(self, password):
"""
Code for password hashing
The hash_password method takes a plain password as argument and stores a hash of it with the user.
This method is called when a new user is registering with the server, or when the user changes the password.
:param password: string
:return: hashed password string
"""
self.password_hash = pwd_context.encrypt(password)
def verify_password(self, password):
"""
The verify_password() method takes a plain password as argument and returns True if the password is
correct or False if not.
This method is called whenever the user provides credentials and they need to be validated.
:param password:
:return: True / False
"""
verification = pwd_context.verify(password, self.password_hash)
return verification
# Method to generate token
def generate_auth_token(self):
s = Serializer(application.config['SECRET_KEY'])
return s.dumps({'id': self.id})
@staticmethod
def verify_auth_token(token):
s = Serializer(application.config['SECRET_KEY'])
try:
data = s.loads(token)
except SignatureExpired:
return None # valid token, but expired
except BadSignature:
return None # invalid token
user = User.query.get(data['id'])
return user
@application.before_request
def before_request():
"""
Add the User object and the db to Flask g.
:return: No return
"""
g.User = User()
g.db = db
"""
Representations
- Adds a response-type into the "Response content type" drop-down menu displayed in Swagger
- When selected, the APP will return the correct response-header and content type
- The default for flask-RESTPlus is application/json
"""
@api.representation('application/xml')
def application_xml(data, code, headers):
resp = representations.xml(data, code, headers)
resp.headers['Content-Type'] = 'application/xml'
return resp
@api.representation('application/json')
def application_json(data, code, headers):
resp = representations.application_json(data, code, headers)
resp.headers['Content-Type'] = 'application/json'
return resp
"""
Error handlers
- exceptions has now been imported from utils!
"""
def log_exception(exception_type):
# We want to know the arguments passed and the path so we can replicate the error
params = dict(request.args)
params['path'] = request.path
# Create the message and log
message = '%s occurred at %s with params=%s' % (exception_type, time.ctime(), params)
logger.exception(message, exc_info=True)
@application.errorhandler(exceptions.RemoteConnectionError)
def remote_connection_error_handler(e):
# Add the Exception to the log ensuring that exc_info is True so that a traceback is also logged
log_exception('RemoteConnectionError')
# Collect Arguments
args = parser.parse_args()
if args['content-type'] != 'application/xml':
return application_json({'message': str(e)},
504,
None)
else:
return application_xml({'message': str(e)},
504,
None)
if __name__ == '__main__':
application.debug = True
application.run(host="127.0.0.1", port=5000)
auth_endpoints.py contains the auth = HTTPBasicAuth(), also the @auth.verify_password decorated function and also imports the auth_functionality_module.py which contains the remaining apiAccess object found in this tutorial
imports
...
from flask_httpauth import HTTPBasicAuth
"""
Auth settings
"""
# Reference
# https://blog.miguelgrinberg.com/post/restful-authentication-with-flask
acc = apiAccess.ApiAccess()
auth = HTTPBasicAuth()
# Password verification decorator
@auth.verify_password
def verify_password(username_or_token, password):
"""
Verifies the passwod or token.
This is the decorates auth method triggered by @auth.login_required
:param username_or_token: string
:param password: string
:return: True / False
"""
# first try to authenticate by token
user = g.User.verify_auth_token(username_or_token)
if not user:
# try to authenticate with username/password
user = g.User.query.filter_by(domain=username_or_token).first()
if not user or not user.verify_password(password):
return False
g.user = user
now = datetime.now().date()
if now > user.expires:
return False
else:
return True
"""
Namespace 1
"""
api1 = Namespace('auth', description='Endpoints for auth control, doc=False)
@api1.route("/token")
@api1.hide
class Token(Resource):
@auth.login_required
@api1.expect(parser, validate=True)
def get(self):
token = g.user.generate_auth_token()
return representations.application_json({'your token': token.decode('ascii')}, 200, None)
"""
Other resources follow as does a second Namespace
The other endpoint.py files in the endpoints directory contain namespaces and endpoints which I want to add @auth.verify_password to. Is there any way to initialise auth
auth = HTTPBasicAuth
and make it accessible and functional to the other namespaces
What I have tried
@application.before_request which gives me RuntimeError: Working outside of application contextMy ideal solution would be to have the auth = HTTPBasicAuth() in app.py and to be able to import it into the endpoint files. If anyone has done this and can provide code or a tutorial that would be great. I find it easier to solve these issues if I can see an example rather than just suggestions please. Thanks
The best way to organize this is (in my opinion) to move the auth object to its own module, let's say authentication.py:
from flask_httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()
@auth.verify_password
def verify_password(username_or_token, password):
# ...
Then you can import the auth object in any of your other modules, where you need to add authentication to an endpoint. For example:
from .authentication import auth
@app.route('/api/whatever')
@auth.login_required
def endpoint():
# ...