#!/usr/bin/env python from flask import Flask, jsonify, request import ipaddress import json app = Flask(__name__) def convert_to_cidr(address, mask): """ Convert an IP address and subnet mask to CIDR notation. Args: address (str): The IP address in dotted decimal format (e.g., '192.168.1.1'). mask (str): The subnet mask in dotted decimal format (e.g., '255.255.255.0'). Returns: str: The CIDR notation of the network (e.g., '192.168.1.1/24'). None: If there is an error in conversion, returns None and prints an error message. Raises: ValueError: If the provided address or mask is invalid. """ # Convertir dirección y máscara a formato CIDR try: # Convertir la máscara de red de formato largo a formato corto # Cada octeto se convierte a su representación binaria y se cuenta el número de bits '1' cidr_mask = ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen red_objeto = f"{address}/{cidr_mask}" return red_objeto except ValueError as e: print(f"Error al convertir a CIDR: {e}") return None subnets_collection = [{"id": 1,"subnet": "192.168.1.0/24", "next-server": "192.168.1.1", "boot-file-name": "pxelinux.0", "reservations": [] }] # Endpoint GET /ogdhcp/v1/status """ Endpoint to get the status of the DHCP service. This endpoint returns a JSON response with the current status of the DHCP service, including disk usage, subnets configuration, and the status of various services. Returns: Response: A JSON response with the following structure: "total": str, # Total disk space "used": str, # Used disk space "available": str, # Available disk space "percentage": str # Percentage of disk space used "id": int, # Subnet ID "subnet": str, # Subnet address "pools": [ "pool": str # IP address pool range "reservations": [ "ip-address": str, # Reserved IP address "hw-address": str # Hardware address associated with the reservation ] ... "kea-ctrl-agent": str, # Status of kea-ctrl-agent service "kea-dhcp4": str, # Status of kea-dhcp4 service "nginx": str # Status of nginx service HTTP Status Code: 200: If the request was successful. """ @app.route('/ogdhcp/v1/status', methods=['GET']) def get_status(): # Simular respuesta de éxito return jsonify({ "success": "Mensaje éxito", "message": { "disk_usage": { "total": "20G", "used": "15G", "available": "5G", "percentage": "75%" }, "subnets": [ { "id": 1, "subnet": "192.168.1.0/24", "pools": [{"pool": "192.168.1.10-192.168.1.100"}], "reservations": [{"ip-address": "192.168.1.20", "hw-address": "00:0c:29:6b:5e:71"}] }, { "id": 2, "subnet": "10.0.0.0/24", "pools": [{"pool": "10.0.0.10-10.0.0.100"}], "reservations": [] } ], "services_status": { "kea-ctrl-agent": "active", "kea-dhcp4": "active", "nginx": "active" } } }), 200 # Endpoint GET /ogdhcp/v1/subnets @app.route('/ogdhcp/v1/subnets', methods=['GET']) def get_subnets(): # Simular respuesta de éxito return jsonify({ "success": "Subredes obtenidas correctamente", "message": subnets_collection }), 200 # Endpoint POST /ogdhcp/v1/subnets """ Create a new subnet. Endpoint: /ogdhcp/v1/subnets Method: POST Request Body (JSON): { "id": , # Unique identifier for the subnet "address": , # IP address of the subnet "mask": , # Subnet mask "nextServer": , # (Optional) Next server IP address "bootFileName": # (Optional) Boot file name } Responses: 200 OK: { "message": } 400 Bad Request: - "Invalid JSON" - "Missing 'id', 'address' or 'mask' key" - { "error": "Error: La subred con el id '' ya existe." } - { "error": "Error: La subred con la dirección '' ya existe." } - "Invalid data format" Description: This endpoint allows the creation of a new subnet. It expects a JSON payload with the subnet details. The function checks if the provided JSON is valid and contains the required keys ('id', 'address', 'mask'). It also ensures that the subnet ID and subnet address do not already exist in the collection. If the validation passes, the new subnet is added to the collection and a success response is returned. """ @app.route('/ogdhcp/v1/subnets', methods=['POST']) def create_subnet(): try: new_subnet = json.loads(request.data) new_subnet["next-server"] = new_subnet.get("nextServer") new_subnet["boot-file-name"] = new_subnet.get("bootFileName") except json.JSONDecodeError: return "Invalid JSON", 400 if isinstance(new_subnet, dict): if "id" in new_subnet and "address" in new_subnet and "mask" in new_subnet: new_subnet_cidr = convert_to_cidr(new_subnet["address"], new_subnet["mask"]) # Comprobar si el id ya existe en la colección for subnet in subnets_collection: if subnet.get("id") == new_subnet["id"]: return jsonify({ "error": f"Error: La subred con el id '{new_subnet['id']}' ya existe." }), 400 if subnet.get("subnet") == new_subnet_cidr: return jsonify({ "error": f"Error: La subred con la dirección '{new_subnet_cidr}' ya existe." }), 400 # Si el id no existe, continuar con la creación de la subred new_subnet["subnet"] = convert_to_cidr(new_subnet["address"], new_subnet["mask"]) subnets_collection.append(new_subnet) return jsonify({ "success": "Subred agregada correctamente", "message": new_subnet }), 200 else: return "Missing 'id', 'address' or 'mask' key", 400 else: return "Invalid data format", 400 # Endpoint DELETE /ogdhcp/v1/subnets/ """ Deletes a subnet from the subnets_collection based on the provided subnetId. Args: subnetId (int): The ID of the subnet to be deleted. Returns: Response: A JSON response indicating the success or failure of the deletion. - If the subnet is successfully deleted, returns a JSON response with a success message and HTTP status 200. - If the subnet with the given ID does not exist, returns a JSON response with an error message and HTTP status 404. """ @app.route('/ogdhcp/v1/subnets/', methods=['DELETE']) def delete_subnet(subnetId): subnet_to_delete = None for subnet in subnets_collection: if int(subnet["id"]) == subnetId: subnet_to_delete = subnet break if subnet_to_delete: print (f"Subnet Collection {subnets_collection}") print (f"Subnet to delete {subnet_to_delete}") subnets_collection.remove(subnet_to_delete) print (f"Subnet Collection {subnets_collection}") return jsonify({ "success": "Subred eliminada correctamente" }), 200 else: return jsonify({ "error": f"Error: La subred con el id '{subnetId}' no existe" }), 404 # Endpoint PUT /ogdhcp/v1/subnets/ """ Updates a subnet with the given subnetId based on the provided JSON data in the request. Args: subnetId (str): The ID of the subnet to be updated. Returns: Response: A JSON response indicating the success or failure of the update operation. - On success: Returns a JSON response with a success message and the updated subnet data, with a status code of 200. - On failure: Returns a JSON response with an error message and a status code of 400. Raises: json.JSONDecodeError: If the request data is not valid JSON. The function performs the following steps: 1. Parses the JSON data from the request. 2. Validates the presence of required fields ('address' and 'mask') in the JSON data. 3. Converts the 'address' and 'mask' fields to a 'subnet' field in CIDR notation. 4. Searches for the subnet with the given subnetId in the subnets_collection. 5. Updates the subnet fields ('subnet', 'nextServer', 'bootFileName') if they are present in the JSON data. 6. Returns a success response if the subnet is found and updated. 7. Returns an error response if the subnet is not found. """ @app.route('/ogdhcp/v1/subnets/', methods=['PUT']) def update_subnet(subnetId): if subnetId == 0: return jsonify({ "error": "Error al guardar la configuración en Kea DHCP: Unable to save configuration" }), 400 try: modify_data = json.loads(request.data) except json.JSONDecodeError: return "Invalid JSON", 400 print ("Modify data", modify_data) print ("Modify data type", type(modify_data)) if modify_data.get("address") != None and modify_data.get("mask") == None: print ("Address", modify_data.get("address")) return jsonify({ "error": f"Error: La máscara de red es requerida con el parametro 'address'" }), 400 if modify_data.get("mask") != None and modify_data.get("address") == None: return jsonify({ "error": f"Error: La dirección de red es requerida con el parametro 'mask'" }), 400 subnet_to_update = None for subnet in subnets_collection: # Casting subnet["id"] to int to avoid type mismatch if str(subnet["id"]) == str(subnetId): # subnet_to_update = subnet if modify_data.get("subnet"): subnet["subnet"] = modify_data["subnet"] if modify_data.get("nextServer"): subnet["next-server"] = modify_data["nextServer"] if modify_data.get("bootFileName"): subnet["boot-file-name"] = modify_data["bootFileName"] subnet_to_update = subnet break if subnet_to_update: # subnet_to_update.update(modify_data) print ("Subnet to update", subnet_to_update) return jsonify({ "success": "Subred modificada correctamente", "message": subnet_to_update }), 200 else: # Si no se encuentra la subred, devolver un error print (f"Subnet Collection {subnets_collection}") response = jsonify({ "error": f"Error: La subred con el id '{subnetId}' no existe" }) response.status_code = 404 response.headers["Content-Type"] = "application/json" return response # Endpoint GET /ogdhcp/v1/subnets//hosts @app.route('/ogdhcp/v1/subnets//hosts', methods=['GET']) def get_hosts(subnetId): # Simular respuesta de éxito return jsonify({ "success": "Hosts retrieved successfully", "message": [ {"ip-address": "192.168.1.10", "hw-address": "00:0c:29:6b:5e:71", "hostname": "host1"}, {"ip-address": "192.168.1.20", "hw-address": "00:0c:29:6b:5e:72", "hostname": "host2"} ] }), 200 # Endpoint POST /ogdhcp/v1/subnets//hosts @app.route('/ogdhcp/v1/subnets//hosts', methods=['POST']) def create_host(subnetId): # Simular respuesta de éxito return jsonify({ "success": "Host agregado correctamente", "message": { "id": 1, "subnet": "192.168.1.0/24", "next-server": "192.168.1.1", "boot-file-name": "pxelinux.0", "reservations": [ {"hostname": "pc11", "hw-address": "56:6f:c7:4f:00:4f", "ip-address": "172.30.4.11"} ] } }), 200 # Endpoint DELETE /ogdhcp/v1/subnets//hosts @app.route('/ogdhcp/v1/subnets//hosts', methods=['DELETE']) def delete_host(subnetId): # Simular respuesta de éxito return jsonify({ "success": "Host eliminado correctamente", "message": { "id": 1, "subnet": "192.168.1.0/24", "next-server": "192.168.1.1", "boot-file-name": "pxelinux.0", "reservations": [ {"hostname": "host2", "hw-address": "00:0c:29:6b:5e:72", "ip-address": "172.30.4.12"} ] } }), 200 # Endpoint PUT /ogdhcp/v1/subnets//hosts @app.route('/ogdhcp/v1/subnets//hosts', methods=['PUT']) def update_host(subnetId): # Simular respuesta de éxito return jsonify({ "success": "Host actualizado correctamente", "message": { "id": 1, "subnet": "192.168.1.0/24", "next-server": "192.168.1.1", "boot-file-name": "pxelinux.0", "reservations": [ {"hostname": "pc11", "hw-address": "56:6f:c7:4f:01:01", "ip-address": "192.168.1.11"}, {"hostname": "host2", "hw-address": "00:0c:29:6b:5e:72", "ip-address": "192.168.1.12"} ] } }), 200 # Endpoint POST /ogdhcp/v1/backup @app.route('/ogdhcp/v1/backup', methods=['POST']) def backup_config(): # Simular respuesta de éxito return jsonify({ "success": "Configuración cargada correctamente" }), 200 if __name__ == '__main__': app.run(debug=True, port=8006)