375 lines
14 KiB
Python
375 lines
14 KiB
Python
#!/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": <str>, # Unique identifier for the subnet
|
|
"address": <str>, # IP address of the subnet
|
|
"mask": <str>, # Subnet mask
|
|
"nextServer": <str>, # (Optional) Next server IP address
|
|
"bootFileName": <str> # (Optional) Boot file name
|
|
}
|
|
Responses:
|
|
200 OK:
|
|
{
|
|
"message": <new_subnet>
|
|
}
|
|
400 Bad Request:
|
|
- "Invalid JSON"
|
|
- "Missing 'id', 'address' or 'mask' key"
|
|
- {
|
|
"error": "Error: La subred con el id '<id>' ya existe."
|
|
}
|
|
- {
|
|
"error": "Error: La subred con la dirección '<subnet>' 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/<subnetId>
|
|
"""
|
|
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/<int:subnetId>', 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/<subnetId>
|
|
"""
|
|
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/<int:subnetId>', 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/<subnetId>/hosts
|
|
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
|
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
|
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
|
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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)
|