Flask Restful template for testing purposes

[rootVIII]

I'm really enjoying flask... lots of fun! Here's a basic example... I use it just for testing things here and there... runs on local loopback and uses a JSON file for storing usernames/passwords.

main.py:

from flask import Flask, jsonify, request, make_response
from flask_restful import Resource, Api
from flask_httpauth import HTTPBasicAuth
from utilities.encrypt import Encrypt
from utilities.decrypt import Decrypt
import json

app = Flask(__name__)
api = Api(app, prefix="/api/v1")
auth = HTTPBasicAuth()


def load_users():
    with open('users.txt') as json_data:
        return json.load(json_data)


def store_users(user_dict):
    with open('users.txt', 'w') as file_out:
        json.dump(user_dict, file_out)


# verifies wherever auth.login_required is present
# False returns 'Unauthorized Access'
@auth.verify_password
def verify(username, password):
    if not (username and password):
        return False
    users = load_users()
    for user, enc_pwd in users.items():
        if user != username:
            continue
        passwd = Decrypt(username).get_plain_txt()
        if passwd != password:
            return False
        return True
    return False


# Login username 'admin' is required for all methods in this class
class PrivateResource1(Resource):
    # get all users
    @auth.login_required
    def get(self):
        if auth.username() != "admin":
            return make_response(jsonify({"Error": "Access Denied"}), 401)
        return jsonify(load_users())

    # add a new username and password to users.txt
    @auth.login_required
    def post(self):
        if auth.username() != "admin":
            return make_response(jsonify({"Error": "Access Denied"}), 401)
        data = request.json
        user, password = next(iter(data.items()))
        users = load_users()
        if user in users:
            return make_response(jsonify({"Error": "User exists"}), 418)
        Encrypt().encrypt_pwd(user, password)
        return jsonify({"Success": "User added"})

    # delete a user
    @auth.login_required
    def delete(self):
        if auth.username() != "admin":
            return make_response(jsonify({"Error": "Access Denied"}), 401)
        user = request.json[0]
        users = load_users()
        try:
            del users[user]
        except Exception:
            return make_response(jsonify({"Error": "User not found"}), 401)
        store_users(users)
        return jsonify({"Success": "%s deleted" % user})

    # modify a user's password
    @auth.login_required
    def put(self):
        if auth.username() != "admin":
            return make_response(jsonify({"Error": "Access Denied"}), 401)
        data = request.json
        user, new_password = next(iter(data.items()))
        users = load_users()
        if user in users:
            Encrypt().encrypt_pwd(user, new_password)
        else:
            return make_response(jsonify({"Error": "User not found"}), 401)
        return jsonify({"Success": "%s updated" % user})


# Simulate a successful REST call and and database query by
# returning user's encrypted password upon authentication...
class PrivateResource2(Resource):
    @auth.login_required
    def get(self, user_name):
        if auth.username() != user_name:
            return make_response(jsonify({"Error": "Access Denied"}), 401)
        users = load_users()
        return jsonify({"Success": "%s" % users[user_name]})


api.add_resource(PrivateResource1, '/users')
api.add_resource(PrivateResource2, '/<string:user_name>')


if __name__ == '__main__':
    app.run(debug=True)

utilities/encrypt.py:

#! /usr/bin/python3
from cryptography.fernet import Fernet
import json
# ensure the correct path to key.txt
# is read in the constructor


class Encrypt:
    def __init__(self):
        with open('key.txt', 'rb') as keyfile:
            self.key = keyfile.read()

    def encrypt_pwd(self, user_name, pwd):
        cipher_suite = Fernet(self.key)
        passwd = pwd.encode()
        cipher_str = cipher_suite.encrypt(passwd)
        self.__write_to_file(user_name, cipher_str.decode('ascii'))

    def __write_to_file(self, user, pwd):
        try:
            with open("users.txt") as json_data1:
                users = json.load(json_data1)
        except Exception:
            users = {}
        users[user] = pwd
        with open("users.txt", 'w') as json_data2:
            json.dump(users, json_data2)

utilities/decrypt.py:

#! /usr/bin/python3
from cryptography.fernet import Fernet
import json


class Decrypt:
    def __init__(self, username):
        with open('key.txt', 'rb') as keyfile:
            self.key = keyfile.read()
        self.plain = self.__decrypt_str(self.__find_match(username))

    def __decrypt_str(self, cipher_str):
        cipher_suite = Fernet(self.key)
        text_bytes = str.encode(cipher_str)
        p_text = cipher_suite.decrypt(text_bytes)
        return p_text.decode("utf-8")

    def __read_in_file(self):
        with open("users.txt") as api_kv_pairs:
            return json.load(api_kv_pairs)

    def __find_match(self, name):
        all_entries = self.__read_in_file()
        for user, pwd in all_entries.items():
            if user == name:
                return all_entries[user]

    def get_plain_txt(self):
        return self.plain