bcgov/mines-digital-trust

Identical blocks of code found in 2 locations. Consider refactoring.
from flask import Flask, jsonify, abort, request, make_response, current_app
Identical blocks of code found in 2 locations. Consider refactoring.
import time, pprint
Identical blocks of code found in 2 locations. Consider refactoring.
from functools import wraps
Identical blocks of code found in 2 locations. Consider refactoring.
from werkzeug.exceptions import Unauthorized
Identical blocks of code found in 2 locations. Consider refactoring.
from app import issuer, logging,app as app_instance
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def secret_key_required(func):
Identical blocks of code found in 2 locations. Consider refactoring.
    @wraps(func)
Identical blocks of code found in 2 locations. Consider refactoring.
    def wrapper(*args, **kwds):
Identical blocks of code found in 2 locations. Consider refactoring.
        if 'ISSUER_SECRET_KEY' not in current_app.ENV:
Identical blocks of code found in 2 locations. Consider refactoring.
            print("NO ISSUER SECRET KEY SET, ALLOWING ALL REQUESTS")
Identical blocks of code found in 2 locations. Consider refactoring.
        if request.headers.get('Issuer-Secret-Key',None) != current_app.ENV['ISSUER_SECRET_KEY']:
Identical blocks of code found in 2 locations. Consider refactoring.
            print(func.__name__ +": NOT_AUTHORIZED")
Identical blocks of code found in 2 locations. Consider refactoring.
            raise Unauthorized('Must know the correct ISSUER_SECRET_KEY to use this method')
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        print(func.__name__ +": AUTHORIZED")
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        return func(*args,**kwds)
Identical blocks of code found in 2 locations. Consider refactoring.
    return wrapper
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
def register_routes(app):
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/health', methods=['GET'])
Identical blocks of code found in 2 locations. Consider refactoring.
    def health_check():
Identical blocks of code found in 2 locations. Consider refactoring.
        if issuer.tob_connection_synced():
Identical blocks of code found in 2 locations. Consider refactoring.
            return make_response(jsonify({'success': True}), 200)
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(503, "Connection not yet synced")
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/readiness', methods=['GET'])
Identical blocks of code found in 2 locations. Consider refactoring.
    def readiness_check():
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        A readiness probe checks if the container is ready to handle requests.
Identical blocks of code found in 2 locations. Consider refactoring.
        A failed readiness probe means that a container should not receive any traffic from a proxy, even if it's running.
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        if issuer.tob_connection_synced():
Identical blocks of code found in 2 locations. Consider refactoring.
            return make_response(jsonify({'success': True}), 200)
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(503, "Connection not ready to process requests")
Identical blocks of code found in 2 locations. Consider refactoring.
   
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/liveness', methods=['GET'])
Identical blocks of code found in 2 locations. Consider refactoring.
    def liveness_check():
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        A liveness probe checks if the container is still running.
Identical blocks of code found in 2 locations. Consider refactoring.
        If the liveness probe fails, the container is killed.
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        if issuer.issuer_liveness_check():
Identical blocks of code found in 2 locations. Consider refactoring.
            return make_response(jsonify({'success': True}), 200)
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(503, "Connection is not live")
Identical blocks of code found in 2 locations. Consider refactoring.
    
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/status/reset', methods=['GET'])
Identical blocks of code found in 2 locations. Consider refactoring.
    @secret_key_required
Identical blocks of code found in 2 locations. Consider refactoring.
    def clear_status():
Identical blocks of code found in 2 locations. Consider refactoring.
        logging.clear_stats()
Identical blocks of code found in 2 locations. Consider refactoring.
        return make_response(jsonify({'success': True}), 200)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/status', methods=['GET'])
Identical blocks of code found in 2 locations. Consider refactoring.
    @secret_key_required
Identical blocks of code found in 2 locations. Consider refactoring.
    def get_status():
Identical blocks of code found in 2 locations. Consider refactoring.
        return make_response(jsonify(logging.get_stats()), 200)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.errorhandler(404)
Identical blocks of code found in 2 locations. Consider refactoring.
    def not_found(error):
Identical blocks of code found in 2 locations. Consider refactoring.
        return make_response(jsonify({'error': 'Not found'}), 404)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/issue-credential', methods=['POST'])
Identical blocks of code found in 2 locations. Consider refactoring.
    @secret_key_required
Identical blocks of code found in 2 locations. Consider refactoring.
    def submit_credential():
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        Exposed method to proxy credential issuance requests.
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        if not issuer.tob_connection_synced():
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(503, "Connection not yet synced")
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        start_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
        method = 'submit_credential.batch'
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        if not request.json:
Identical blocks of code found in 2 locations. Consider refactoring.
            end_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
            logging.log_timing_method(method, start_time, end_time, False)
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(400)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        cred_input = request.json
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        response = issuer.handle_send_credential(cred_input)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        end_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
        logging.log_timing_method(method, start_time, end_time, True)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        return response
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
    #not protected, used by ACA-py... TODO enforce this.
Identical blocks of code found in 2 locations. Consider refactoring.
    @app.route('/api/agentcb/topic/<topic>/', methods=['POST'])
Identical blocks of code found in 2 locations. Consider refactoring.
    def agent_callback(topic):
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        Main callback for aries agent.  Dispatches calls based on the supplied topic.
Identical blocks of code found in 2 locations. Consider refactoring.
        """
Identical blocks of code found in 2 locations. Consider refactoring.
        start_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
        method = 'agent_callback.' + topic
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        if not request.json:
Identical blocks of code found in 2 locations. Consider refactoring.
            end_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
            logging.log_timing_method(method, start_time, end_time, False)
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(400)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        message = request.json
Identical blocks of code found in 2 locations. Consider refactoring.
        logging.log_timing_event(method, message, start_time, None, False)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        # dispatch based on the topic type
Identical blocks of code found in 2 locations. Consider refactoring.
        if topic == issuer.TOPIC_CONNECTIONS:
Identical blocks of code found in 2 locations. Consider refactoring.
            if "state" in message:
Identical blocks of code found in 2 locations. Consider refactoring.
                method = method + '.' + message["state"]
Identical blocks of code found in 2 locations. Consider refactoring.
                response = issuer.handle_connections(message["state"], message)
Identical blocks of code found in 2 locations. Consider refactoring.
            else:
Identical blocks of code found in 2 locations. Consider refactoring.
                response = jsonify({})
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_CONNECTIONS_ACTIVITY:
Identical blocks of code found in 2 locations. Consider refactoring.
            response = jsonify({})
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_CREDENTIALS:
Identical blocks of code found in 2 locations. Consider refactoring.
            if "state" in message:
Identical blocks of code found in 2 locations. Consider refactoring.
                method = method + '.' + message["state"]
Identical blocks of code found in 2 locations. Consider refactoring.
                response = issuer.handle_credentials(message["state"], message)
Identical blocks of code found in 2 locations. Consider refactoring.
            else:
Identical blocks of code found in 2 locations. Consider refactoring.
                response = jsonify({})
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_PRESENTATIONS:
Identical blocks of code found in 2 locations. Consider refactoring.
            if "state" in message:
Identical blocks of code found in 2 locations. Consider refactoring.
                method = method + '.' + message["state"]
Identical blocks of code found in 2 locations. Consider refactoring.
                response = issuer.handle_presentations(message["state"], message)
Identical blocks of code found in 2 locations. Consider refactoring.
            else:
Identical blocks of code found in 2 locations. Consider refactoring.
                response = jsonify({})
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_GET_ACTIVE_MENU:
Identical blocks of code found in 2 locations. Consider refactoring.
            response = issuer.handle_get_active_menu(message)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_PERFORM_MENU_ACTION:
Identical blocks of code found in 2 locations. Consider refactoring.
            response = issuer.handle_perform_menu_action(message)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_ISSUER_REGISTRATION:
Identical blocks of code found in 2 locations. Consider refactoring.
            response = issuer.handle_register_issuer(message)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        elif topic == issuer.TOPIC_PROBLEM_REPORT:
Identical blocks of code found in 2 locations. Consider refactoring.
            response = issuer.handle_problem_report(message)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        else:
Identical blocks of code found in 2 locations. Consider refactoring.
            print("Callback: topic=", topic, ", message=", message)
Identical blocks of code found in 2 locations. Consider refactoring.
            end_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
            logging.log_timing_method(method, start_time, end_time, False)
Identical blocks of code found in 2 locations. Consider refactoring.
            logging.log_timing_event(method, message, start_time, end_time, False)
Identical blocks of code found in 2 locations. Consider refactoring.
            abort(400, {'message': 'Invalid topic: ' + topic})
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        end_time = time.perf_counter()
Identical blocks of code found in 2 locations. Consider refactoring.
        logging.log_timing_method(method, start_time, end_time, True)
Identical blocks of code found in 2 locations. Consider refactoring.
        logging.log_timing_event(method, message, start_time, end_time, True)
Identical blocks of code found in 2 locations. Consider refactoring.
 
Identical blocks of code found in 2 locations. Consider refactoring.
        return response