Unverified Commit 4bd5e02e authored by Guillermo Sanz López's avatar Guillermo Sanz López Committed by GitHub
Browse files

Merge pull request #8 from Telefonica/env-variable-certificates 

Env variable certificates
parents 9828322f adb920ec

capif_backend/backoffice_service/__main__.py

+83 −56
Original line number Diff line number Diff line 
from flask import Flask

from flask import Flask, request, make_response

from OpenSSL.crypto import PKey, TYPE_RSA, X509Req, dump_certificate_request, FILETYPE_PEM, dump_privatekey

from flask_jwt_extended import JWTManager

from .configs.config import Config
@@ -11,36 +11,42 @@ from .controllers.providers_controller import providers_routes 
from .controllers.users_controller import users_routes

from .controllers.apis_controller import apis_routes

from .controllers.login import access_routes

from .controllers.configuration_controller import configuration_routes

from flask_cors import CORS



# Create Flask app

app = Flask(__name__)



# Configurar CORS para todas las rutas

CORS(app, resources={r"/api/*": {"origins": "http://localhost:3001"}})

# Configurar CORS correctamente

CORS(app, supports_credentials=True, allow_headers=["Content-Type", "Authorization"])



#CORS(app) # habilita CORS para todas las rutas en la aplicación

# Manejo global de preflight OPTIONS

@app.before_request

def handle_preflight():

    if request.method == "OPTIONS":

        res = make_response("", 200)

        res.headers["Access-Control-Allow-Origin"] = "*"

        res.headers["Access-Control-Allow-Methods"] = "GET, POST, PATCH, PUT, DELETE, OPTIONS"

        res.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"

        return res



# Register blueprints (routes)

# other option modify URL ->  app.register_blueprint(backoffice_routes, url_prefix="/resources")

app.register_blueprint(backoffice_routes, url_prefix="/api")

app.register_blueprint(access_routes, url_prefix="/api")

app.register_blueprint(invokers_routes, url_prefix="/api/invokers")

app.register_blueprint(providers_routes, url_prefix="/api/providers")

app.register_blueprint(apis_routes, url_prefix="/api/apis")

app.register_blueprint(users_routes, url_prefix="/api/users")



app.register_blueprint(configuration_routes, url_prefix="/api")



@app.route("/api/options", methods=["OPTIONS"])

def handle_options():

    return "", 200





# @app.route('/', defaults={'path': ''})

# @app.route('/<path:path>', methods=['OPTIONS'])

# def handle_options(path):     

#     return 'Success', 200



# @app.route("/api/configuration", methods=["OPTIONS"])

# def handle_options_conf():

#     return "", 200



# Configure JWT

app.config["JWT_SECRET_KEY"] = "super-secret"
@@ -68,11 +74,19 @@ csr_request = dump_certificate_request(FILETYPE_PEM, req) 
# Get private key

private_key = dump_privatekey(FILETYPE_PEM, key)



# Leer variable de entorno para determinar si se carga desde Vault o desde variables de entorno

LOAD_FROM_VAULT = os.getenv('LOAD_FROM_VAULT', 'true').lower() == 'true'



# Ensure the certs directory exists

os.makedirs('newcerts', exist_ok=True)



if LOAD_FROM_VAULT:

    # Cargar certificados desde Vault

    print(f"Using Vault token: {os.getenv('VAULT_TOKEN')}")



# Get ca.cert from vault

    # Get ca.cert from Vault

    config = Config().get_config()  # get configuration from config.py

url = 'http://{}:{}/v1/secret/data/ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))   # http://vault:8200/v1/secret/data/ca

    url = 'http://{}:{}/v1/secret/data/ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))  

    headers = {'X-Vault-Token': os.getenv('VAULT_TOKEN')}

    response = requests.request("GET", url, headers=headers, verify=False)
@@ -82,14 +96,13 @@ print(response) 


    ca_cert = json.loads(response.text)["data"]["data"]["ca"]



# print ("CA CERT URL: ", url)

# print ("VAULT_HOSTNAME", os.get_environ('VAULT_HOSTNAME'))

# print ("VAULT_PORT", os.get_environ('VAULT_PORT'))



# Request Vault a CA certificate to backoffice application

url = 'http://{}:{}/v1/pki_int/sign/my-ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))  # http://vault:8200/v1/pki_int/sign/my-ca

    # Save ca_cert

    ca_file = open('newcerts/ca_root.crt', 'wb+')

    ca_file.write(bytes(ca_cert, 'utf-8'))

    ca_file.close()



headers = {'X-Vault-Token': os.getenv('VAULT_TOKEN')}  # token = dev-only-token -> for dev environment

    # Request Vault to sign the CSR

    url = 'http://{}:{}/v1/pki_int/sign/my-ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))  

    data = {

        'format': 'pem_bundle',

        'ttl': '43000h',
@@ -99,15 +112,6 @@ data = { 
    response = requests.request("POST", url, headers=headers, data=data, verify=False)

    backoffice_cert = json.loads(response.text)['data']['certificate']





# Ensure the certs directory exists

os.makedirs('newcerts', exist_ok=True)



# Save ca_cert

ca_file = open('newcerts/ca_root.crt', 'wb+')

ca_file.write(bytes(ca_cert, 'utf-8'))

ca_file.close()



    # Save private_key

    private_key_file = open("newcerts/superadmin.key", 'wb+')

    private_key_file.write(private_key)
@@ -118,10 +122,33 @@ certification_file = open('newcerts/superadmin.crt', 'wb+') 
    certification_file.write(bytes(backoffice_cert, 'utf-8'))

    certification_file.close()



else:

    # Cargar certificados desde variables de entorno

    ca_cert = os.getenv('CA_CERT')

    private_key_content = os.getenv('PRIVATE_KEY')

    backoffice_cert_content = os.getenv('BACKOFFICE_CERT')



    if not ca_cert or not private_key_content or not backoffice_cert_content:

        print("Error: Las variables de entorno CA_CERT, PRIVATE_KEY y BACKOFFICE_CERT deben estar definidas")

        exit(1)



    # Guardar el certificado CA

    with open('newcerts/ca_root.crt', 'wb+') as ca_file:

        ca_file.write(ca_cert.replace('\\n', '\n').encode('utf-8'))



    # Guardar la clave privada

    with open('newcerts/superadmin.key', 'wb+') as private_key_file:

        private_key_file.write(private_key_content.replace('\\n', '\n').encode('utf-8'))



    # Guardar el certificado de backoffice

    with open('newcerts/superadmin.crt', 'wb+') as certification_file:

        certification_file.write(backoffice_cert_content.replace('\\n', '\n').encode('utf-8'))





#----------------------------------------

# launch

#----------------------------------------



# the applicacions starts here in every interface and port 8080

# The application starts here, on every interface and port 8080

if __name__ == "__main__":

    app.run(debug=True, host='0.0.0.0', port=8080)

capif_backend/backoffice_service/controllers/configuration_controller.py

0 → 100644
+225 −0
Original line number Diff line number Diff line 
#!/usr/bin/env python3



from flask import Blueprint, jsonify

from ..core.operations import Operations

from flask_jwt_extended import jwt_required, get_jwt_identity

from flask_cors import cross_origin

import requests

from flask import request, make_response

import jsonschema

from jsonschema import validate, ValidationError

from ..utils.utils import is_snake_case, validate_snake_case_keys





# create blueprint backoffice_routes

configuration_routes = Blueprint("configuration_routes", __name__)

operations = Operations()   # create operations object -> operations.py





@cross_origin()

@configuration_routes.route("/configuration", methods=["GET"])

@jwt_required()

def get_configuration():

    return operations.get_configuration()





@configuration_routes.route("/configuration/update-param", methods=["PATCH", "OPTIONS"])

@cross_origin()

@jwt_required()

def update_configuration_param():

    if request.method == "OPTIONS":

        return make_response("", 200)

    

    data = request.json

    param_path = data.get("param_path")

    new_value = data.get("new_value")



    if not param_path or new_value is None:

        return jsonify({"error": "param_path y new_value son requeridos"}), 400



    return operations.update_configuration_param(param_path, new_value)





@configuration_routes.route("/configuration/replace-config", methods=["PUT"])

@cross_origin()

@jwt_required()

def replace_configuration():

    """ Reemplaza toda la configuración con una nueva, verificando su estructura """

    new_config = request.json



    if not new_config:

        return jsonify({"error": "La nueva configuración no puede estar vacía"}), 400



    # Verificar estructura general

    required_keys = ["config_name", "version", "description", "settings"]

    for key in required_keys:

        if key not in new_config:

            return jsonify({"error": f"Falta la clave requerida: {key}"}), 400



    if not isinstance(new_config["settings"], dict) or not new_config["settings"]:

        return jsonify({"error": "'settings' debe ser un objeto con al menos una categoría"}), 400



    error_response = validate_snake_case_keys(new_config["settings"])

    if error_response:

        return error_response  # Retorna el error si se encuentra una clave incorrecta



    return operations.replace_configuration(new_config)





@configuration_routes.route("/configuration/add-category", methods=["POST"])

@cross_origin()

@jwt_required()

def add_new_category():

    """Añade una nueva categoría dentro de 'settings'."""

    data = request.json

    category_name = data.get("category_name")

    category_values = data.get("category_values")



    if not category_name or not category_values:

        return jsonify({"error": "category_name y category_values son requeridos"}), 400

    

    return operations.add_new_category(category_name, category_values)





@configuration_routes.route("/configuration/add-config-setting", methods=["PATCH"])

@cross_origin()

@jwt_required()

def add_new_config_setting():

    """Añade un nuevo parámetro dentro de una categoría en 'settings'."""

    data = request.json

    param_path = data.get("param_path")

    new_value = data.get("new_value")

    

    if not param_path or new_value is None:

        return jsonify({"error": "param_path y new_value son requeridos"}), 400

    

    return operations.add_new_config_setting(param_path, new_value)





@configuration_routes.route("/configuration/remove-config-param", methods=["DELETE", "OPTIONS"])

@cross_origin()

@jwt_required()

def remove_configuration_param():

    if request.method == "OPTIONS":

        return make_response("", 200)



    data = request.json

    param_path = data.get("param_path")



    if not param_path:

        return jsonify({"error": "param_path es requerido"}), 400



    return operations.remove_configuration_param(param_path)





@configuration_routes.route("/configuration/remove-config-category", methods=["DELETE", "OPTIONS"])

@cross_origin()

@jwt_required()

def remove_configuration_category():

    if request.method == "OPTIONS":

        return make_response("", 200)



    data = request.json

    category_name = data.get("category_name")



    if not category_name:

        return jsonify({"error": "category_name es requerido"}), 400



    return operations.remove_configuration_category(category_name)







### REGISTER CONFIGURATION ###



@cross_origin()

@configuration_routes.route("/configuration/register", methods=["GET"])

@jwt_required()

def get_register_configuration():

    """ Obtiene la configuración del register """

    return operations.get_register_configuration()





@cross_origin()

@configuration_routes.route("/configuration/register/update-param", methods=["PATCH"])

@jwt_required()

def update_register_configuration_param():

    """ Actualiza un único parámetro en la configuración del register """

    data = request.json

    param_path = data.get("param_path")

    new_value = data.get("new_value")



    if not param_path or new_value is None:

        return jsonify({"error": "param_path y new_value son requeridos"}), 400



    return operations.update_register_configuration_param(param_path, new_value)





@cross_origin()

@configuration_routes.route("/configuration/register/replace-config", methods=["PUT"])

@jwt_required()

def replace_register_configuration():

    """ Reemplaza toda la configuración del register con una nueva """

    new_config = request.json



    if not new_config:

        return jsonify({"error": "La nueva configuración no puede estar vacía"}), 400



    return operations.replace_register_configuration(new_config)





@cross_origin()

@configuration_routes.route("/configuration/register/add-category-config", methods=["POST"])

@jwt_required()

def add_new_config_regsiter_category():

    """Añade una nueva categoría dentro de 'settings' en el register."""

    data = request.json

    category_name = data.get("category_name")

    category_values = data.get("category_values")



    if not category_name or not category_values:

        return jsonify({"error": "category_name y category_values son requeridos"}), 400

    

    return operations.add_new_config_register_category(category_name, category_values)





@cross_origin()

@configuration_routes.route("/configuration/register/add-config-param-setting", methods=["PATCH"])

@jwt_required()

def add_new_config_register_setting():

    """Añade un nuevo parámetro dentro de una categoría en 'settings' en el register."""

    data = request.json

    param_path = data.get("param_path")

    new_value = data.get("new_value")

    

    if not param_path or new_value is None:

        return jsonify({"error": "param_path y new_value son requeridos"}), 400

    

    return operations.add_new_config_register_setting(param_path, new_value)





@cross_origin()

@configuration_routes.route("/configuration/register/remove-config-param", methods=["DELETE"])

@jwt_required()

def remove_register_configuration_param():

    """Elimina un parámetro dentro de una categoría en 'settings' en el register."""

    data = request.json

    param_path = data.get("param_path")



    if not param_path:

        return jsonify({"error": "param_path es requerido"}), 400



    return operations.remove_register_configuration_param(param_path)





@cross_origin()

@configuration_routes.route("/configuration/register/remove-config-category", methods=["DELETE"])

@jwt_required()

def remove_register_configuration_category():

    """Elimina una categoría completa en 'settings' en el register."""

    data = request.json

    category_name = data.get("category_name")



    if not category_name:

        return jsonify({"error": "category_name es requerido"}), 400



    return operations.remove_register_configuration_category(category_name)

capif_backend/backoffice_service/controllers/invokers_controller.py

+0 −25
Original line number Diff line number Diff line
@@ -115,31 +115,6 @@ def delete_invokers_route(): 
def get_invokers():

    return operations.get_invokers()



# GET invokers with pagination

@cross_origin()

@invokers_routes.route("/pages", methods=["GET"])

@jwt_required()

def get_invokersPages():

    page = request.args.get('page', default=1, type=int)

    page_size = request.args.get('pageSize', default=10, type=int)

    return operations.get_invokersPages(page, page_size)



# SEARCH invokers

@cross_origin()

@invokers_routes.route("/search", methods=["GET"])

@jwt_required()

def search_invokers():

    try:

        search_term = request.args.get('searchTerm', default='', type=str)

        page = request.args.get('page', default=1, type=int)

        page_size = request.args.get('pageSize', default=10, type=int)



        response, status_code = operations.search_invokers(search_term, page, page_size)



        return jsonify(response), status_code

    except Exception as e:

        return jsonify({"error": str(e)}), 500



# GET the number of invokers

@cross_origin()

@invokers_routes.route("/numInvokers", methods=["GET"])

capif_backend/backoffice_service/core/operations.py

+577 −2

File changed.

Preview size limit exceeded, changes collapsed.

capif_backend/backoffice_service/utils/utils.py

0 → 100644
+17 −0
Original line number Diff line number Diff line 
import re

from flask import jsonify



def is_snake_case(value):

    """Verifica si una clave está en snake_case (solo letras minúsculas, números y guiones bajos)."""

    return bool(re.match(r'^[a-z0-9_]+$', value))



# Verificar que TODAS las claves dentro de settings estén en snake_case

def validate_snake_case_keys(obj, path="settings"):

    """Recorre el JSON validando que todas las claves estén en snake_case."""

    for key, value in obj.items():

        if not is_snake_case(key):

            return jsonify({"error": f"La clave '{path}.{key}' no está en snake_case"}), 400

        if isinstance(value, dict):

            error_response = validate_snake_case_keys(value, f"{path}.{key}")

            if error_response:

                return error_response 
 No newline at end of file