thirdparty
/
CybroAddons
mirror of https://github.com/CybroOdoo/CybroAddons.git
5
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Improvements in authentication and structure

pull/401/head
Bernat Roig 2 weeks ago
committed by GitHub
parent
ff60d65070
commit
92d2823a53
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
1 changed files with 364 additions and 316 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 680
      rest_api_odoo/controllers/rest_api_odoo.py

680
rest_api_odoo/controllers/rest_api_odoo.py
View File

@ -5,6 +5,7 @@
# #
# Copyright (C) 2024-TODAY Cybrosys Technologies(<https://www.cybrosys.com>) # Copyright (C) 2024-TODAY Cybrosys Technologies(<https://www.cybrosys.com>)
# Author: Ayana KP (odoo@cybrosys.com) # Author: Ayana KP (odoo@cybrosys.com)
# Modified by: Broigm - Improvements in authentication and structure
# #
# You can modify it under the terms of the GNU LESSER # You can modify it under the terms of the GNU LESSER
# GENERAL PUBLIC LICENSE (LGPL v3), Version 3. # GENERAL PUBLIC LICENSE (LGPL v3), Version 3.
@ -22,347 +23,394 @@
import json import json
import logging import logging
import base64 import base64
from datetime import datetime, date, timedelta
from odoo import http from odoo import http
from odoo.http import request from odoo.http import request
from datetime import datetime, date from werkzeug.exceptions import BadRequest, Unauthorized, NotFound, MethodNotAllowed
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
class RestApi(http.Controller): class RestApi(http.Controller):
"""This is a controller which is used to generate responses based on the """Controlador API REST mejorado con autenticación basada en API Key solamente"""
api requests"""
def _json_response(self, data, status=200):
def auth_api_key(self, api_key): """Genera respuesta JSON estandarizada"""
"""This function is used to authenticate the api-key when sending a response = request.make_response(
request""" json.dumps(data, ensure_ascii=False, indent=2),
user_id = request.env['res.users'].sudo().search([('api_key', '=', api_key)]) headers=[('Content-Type', 'application/json')]
if api_key is not None and user_id: )
response = True response.status_code = status
elif not user_id:
response = ('<html><body><h2>Invalid <i>API Key</i> '
'!</h2></body></html>')
else:
response = ("<html><body><h2>No <i>API Key</i> Provided "
"!</h2></body></html>")
return response return response
def generate_response(self, method, model, rec_id, request_params=None): def _error_response(self, message, status=400, error_code=None):
"""This function is used to generate the response based on the type """Genera respuesta de error estandarizada"""
of request and the parameters given""" error_data = {
option = request.env['connection.api'].search( 'error': True,
[('model_id', '=', model)], limit=1) 'message': message,
model_name = option.model_id.model 'status_code': status
}
if error_code:
error_data['error_code'] = error_code
return self._json_response(error_data, status)
def _authenticate_api_key(self, api_key):
"""
Autentica usando solo la API key y configura la sesión del usuario
Returns: (success: bool, user_id: int or None, error_message: str or None)
"""
if not api_key:
return False, None, "API Key no proporcionada"
try:
user = request.env['res.users'].sudo().search([
('api_key', '=', api_key),
('active', '=', True)
], limit=1)
if not user:
return False, None, "API Key inválida o usuario inactivo"
# Verificar si la API key no ha expirado (si implementas expiración)
if hasattr(user, 'api_key_expiry') and user.api_key_expiry:
if user.api_key_expiry < datetime.now():
return False, None, "API Key expirada"
# Configurar la sesión con el usuario autenticado
request.session.uid = user.id
request.env.user = user
return True, user.id, None
except Exception as e:
_logger.error(f"Error en autenticación API: {str(e)}")
return False, None, "Error interno de autenticación"
def _serialize_record_values(self, records):
"""Serializa los valores de los registros para JSON"""
if not records:
return []
serialized_records = []
for record in records:
serialized_record = {}
for key, value in record.items():
if isinstance(value, (datetime, date)):
serialized_record[key] = value.isoformat()
elif isinstance(value, bytes):
serialized_record[key] = base64.b64encode(value).decode('utf-8')
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)):
try:
serialized_record[key] = list(value) if value else []
except:
serialized_record[key] = str(value)
else:
serialized_record[key] = value
serialized_records.append(serialized_record)
return serialized_records
def _get_model_config(self, model_name):
"""Obtiene la configuración de la API para un modelo"""
model_obj = request.env['ir.model'].sudo().search([('model', '=', model_name)], limit=1)
if not model_obj:
return None, "Modelo no encontrado"
api_config = request.env['connection.api'].sudo().search([
('model_id', '=', model_obj.id)
], limit=1)
if not api_config:
return None, "Modelo no configurado para API REST"
return api_config, None
def _parse_request_data(self, method):
"""Parsea los datos de la request según el método HTTP"""
data = {}
fields = []
# Handle data based on method
if method == 'GET': if method == 'GET':
# For GET requests, check both JSON body and query parameters # Para GET, intentar obtener campos del query string o JSON body
data = request_params or {} query_params = dict(request.httprequest.args)
fields = []
# First, try to get fields from JSON body
try: try:
if request.httprequest.data: if request.httprequest.data:
json_data = json.loads(request.httprequest.data) json_data = json.loads(request.httprequest.data)
data.update(json_data)
if 'fields' in json_data: if 'fields' in json_data:
fields = json_data['fields'] fields = json_data['fields']
except json.JSONDecodeError: except json.JSONDecodeError:
pass # If JSON is invalid, continue with query params pass
# If no fields from JSON, try query parameters if not fields and 'fields' in query_params:
if not fields: fields = [field.strip() for field in query_params['fields'].split(',')]
fields_param = data.get('fields', '')
if fields_param: elif method in ['POST', 'PUT']:
fields = [field.strip() for field in fields_param.split(',')]
# If still no fields, use defaults based on record type
if not fields:
if rec_id != 0:
# For specific record, get all fields
fields = None # This will get all available fields
else:
# For all records, use minimal fields
fields = ['id', 'display_name']
elif method != 'DELETE':
# For POST/PUT requests, parse JSON from body
try: try:
if request.httprequest.data: if request.httprequest.data:
data = json.loads(request.httprequest.data) data = json.loads(request.httprequest.data)
if 'fields' in data:
fields = data['fields']
else: else:
data = {} return None, None, "No se proporcionaron datos JSON"
except json.JSONDecodeError: except json.JSONDecodeError:
return ("<html><body><h2>Invalid JSON Data" return None, None, "JSON inválido"
"</h2></body></html>")
else: return data, fields, None
# DELETE method
data = {} @http.route(['/api/v1/auth'], type='http', auth='none', methods=['POST'], csrf=False)
fields = [] def authenticate(self, **kw):
"""Endpoint de autenticación que genera API key"""
# Extract fields for POST/PUT methods
if method in ['POST', 'PUT'] and data:
fields = []
if 'fields' in data:
for field in data['fields']:
fields.append(field)
if not fields and method != 'DELETE' and method != 'GET':
return ("<html><body><h2>No fields selected for the model"
"</h2></body></html>")
if not option:
return ("<html><body><h2>No Record Created for the model"
"</h2></body></html>")
try: try:
if method == 'GET': data = json.loads(request.httprequest.data or '{}')
if not option.is_get: except json.JSONDecodeError:
return ("<html><body><h2>Method Not Allowed" return self._error_response("JSON inválido", 400)
"</h2></body></html>")
else: username = data.get('username') or request.httprequest.headers.get('username')
datas = [] password = data.get('password') or request.httprequest.headers.get('password')
if rec_id != 0: database = data.get('database') or request.httprequest.headers.get('database', request.env.cr.dbname)
# For specific record
search_fields = fields if fields is not None else [] if not all([username, password]):
partner_records = request.env[ return self._error_response("Username y password son requeridos", 400)
str(model_name)].search_read(
domain=[('id', '=', rec_id)], try:
fields=search_fields # Actualizar sesión con la base de datos
) request.session.update(http.get_default_session(), db=database)
for record in partner_records:
for key, value in record.items(): # Autenticar credenciales
if isinstance(value, (datetime, date)): auth_result = request.session.authenticate(
record[key] = value.isoformat() database,
elif isinstance(value, bytes): {'login': username, 'password': password, 'type': 'password'}
# Convert bytes to base64 string for JSON serialization )
import base64
record[key] = base64.b64encode(value).decode('utf-8') if not auth_result:
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)): return self._error_response("Credenciales inválidas", 401)
# Handle other non-serializable iterables
try: # Generar o recuperar API key
record[key] = list(value) if value else [] user = request.env['res.users'].browse(auth_result['uid'])
except: api_key = user.generate_api_key()
record[key] = str(value)
elif isinstance(value, bytes): response_data = {
# Convert bytes to base64 string for JSON serialization "success": True,
import base64 "message": "Autenticación exitosa",
record[key] = base64.b64encode(value).decode('utf-8') "data": {
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)): "user_id": user.id,
# Handle other non-serializable iterables "username": user.login,
try: "name": user.name,
record[key] = list(value) if value else [] "api_key": api_key,
except: "database": database
record[key] = str(value) }
data = json.dumps({ }
'records': partner_records
}) return self._json_response(response_data)
datas.append(data)
return request.make_response(data=datas)
else:
# For all records
search_fields = fields if fields is not None else ['id', 'display_name']
partner_records = request.env[
str(model_name)].search_read(
domain=[],
fields=search_fields
)
for record in partner_records:
for key, value in record.items():
if isinstance(value, (datetime, date)):
record[key] = value.isoformat()
data = json.dumps({
'records': partner_records
})
datas.append(data)
return request.make_response(data=datas)
except Exception as e: except Exception as e:
_logger.error(f"Error in GET method: {str(e)}") _logger.error(f"Error en autenticación: {str(e)}")
return ("<html><body><h2>Error processing request" return self._error_response("Error interno de autenticación", 500)
"</h2></body></html>")
@http.route(['/api/v1/<model_name>', '/api/v1/<model_name>/<int:record_id>'],
if method == 'POST': type='http', auth='none', methods=['GET', 'POST', 'PUT', 'DELETE'], csrf=False)
if not option.is_post: def api_handler(self, model_name, record_id=None, **kw):
return ("<html><body><h2>Method Not Allowed" """Endpoint principal de la API REST"""
"</h2></body></html>") method = request.httprequest.method
else:
try: # Autenticación usando API key
datas = [] api_key = request.httprequest.headers.get('X-API-Key') or request.httprequest.headers.get('api-key')
new_resource = request.env[str(model_name)].create( success, user_id, error_msg = self._authenticate_api_key(api_key)
data['values'])
partner_records = request.env[ if not success:
str(model_name)].search_read( return self._error_response(error_msg, 401, "AUTHENTICATION_FAILED")
domain=[('id', '=', new_resource.id)],
fields=fields # Obtener configuración del modelo
) api_config, error_msg = self._get_model_config(model_name)
for record in partner_records: if not api_config:
for key, value in record.items(): return self._error_response(error_msg, 404, "MODEL_NOT_CONFIGURED")
if isinstance(value, (datetime, date)):
record[key] = value.isoformat() # Verificar permisos del método
elif isinstance(value, bytes): method_permissions = {
# Convert bytes to base64 string for JSON serialization 'GET': api_config.is_get,
import base64 'POST': api_config.is_post,
record[key] = base64.b64encode(value).decode('utf-8') 'PUT': api_config.is_put,
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)): 'DELETE': api_config.is_delete
# Handle other non-serializable iterables }
try:
record[key] = list(value) if value else [] if not method_permissions.get(method, False):
except: return self._error_response(f"Método {method} no permitido para este modelo", 405, "METHOD_NOT_ALLOWED")
record[key] = str(value)
elif isinstance(value, bytes): # Parsear datos de la request
# Convert bytes to base64 string for JSON serialization data, fields, error_msg = self._parse_request_data(method)
import base64 if error_msg:
record[key] = base64.b64encode(value).decode('utf-8') return self._error_response(error_msg, 400, "INVALID_REQUEST_DATA")
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)):
# Handle other non-serializable iterables try:
try: return self._handle_request(method, api_config.model_id.model, record_id, data, fields)
record[key] = list(value) if value else [] except Exception as e:
except: _logger.error(f"Error procesando request {method} para {model_name}: {str(e)}")
record[key] = str(value) return self._error_response("Error interno del servidor", 500, "INTERNAL_SERVER_ERROR")
new_data = json.dumps({'New resource': partner_records, })
datas.append(new_data) def _handle_request(self, method, model_name, record_id, data, fields):
return request.make_response(data=datas) """Maneja las diferentes operaciones CRUD"""
except Exception as e: model = request.env[model_name]
_logger.error(f"Error in POST method: {str(e)}")
return ("<html><body><h2>Invalid JSON Data" if method == 'GET':
"</h2></body></html>") return self._handle_get(model, record_id, fields)
elif method == 'POST':
if method == 'PUT': return self._handle_post(model, data, fields)
if not option.is_put: elif method == 'PUT':
return ("<html><body><h2>Method Not Allowed" return self._handle_put(model, record_id, data, fields)
"</h2></body></html>") elif method == 'DELETE':
else: return self._handle_delete(model, record_id)
if rec_id == 0:
return ("<html><body><h2>No ID Provided" def _handle_get(self, model, record_id, fields):
"</h2></body></html>") """Maneja requests GET"""
else: try:
resource = request.env[str(model_name)].browse( if record_id:
int(rec_id)) # Obtener registro específico
if not resource.exists(): domain = [('id', '=', record_id)]
return ("<html><body><h2>Resource not found" search_fields = fields if fields else []
"</h2></body></html>")
else:
try:
datas = []
resource.write(data['values'])
partner_records = request.env[
str(model_name)].search_read(
domain=[('id', '=', resource.id)],
fields=fields
)
for record in partner_records:
for key, value in record.items():
if isinstance(value, (datetime, date)):
record[key] = value.isoformat()
elif isinstance(value, bytes):
# Convert bytes to base64 string for JSON serialization
import base64
record[key] = base64.b64encode(value).decode('utf-8')
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)):
# Handle other non-serializable iterables
try:
record[key] = list(value) if value else []
except:
record[key] = str(value)
elif isinstance(value, bytes):
# Convert bytes to base64 string for JSON serialization
import base64
record[key] = base64.b64encode(value).decode('utf-8')
elif hasattr(value, '__iter__') and not isinstance(value, (str, dict)):
# Handle other non-serializable iterables
try:
record[key] = list(value) if value else []
except:
record[key] = str(value)
new_data = json.dumps(
{'Updated resource': partner_records,
})
datas.append(new_data)
return request.make_response(data=datas)
except Exception as e:
_logger.error(f"Error in PUT method: {str(e)}")
return ("<html><body><h2>Invalid JSON Data "
"!</h2></body></html>")
if method == 'DELETE':
if not option.is_delete:
return ("<html><body><h2>Method Not Allowed"
"</h2></body></html>")
else:
if rec_id == 0:
return ("<html><body><h2>No ID Provided"
"</h2></body></html>")
else:
resource = request.env[str(model_name)].browse(
int(rec_id))
if not resource.exists():
return ("<html><body><h2>Resource not found"
"</h2></body></html>")
else:
records = request.env[
str(model_name)].search_read(
domain=[('id', '=', resource.id)],
fields=['id', 'display_name']
)
remove = json.dumps(
{"Resource deleted": records,
})
resource.unlink()
return request.make_response(data=remove)
@http.route(['/send_request'], type='http',
auth='none',
methods=['GET', 'POST', 'PUT', 'DELETE'], csrf=False)
def fetch_data(self, **kw):
"""This controller will be called when sending a request to the
specified url, and it will authenticate the api-key and then will
generate the result"""
http_method = request.httprequest.method
api_key = request.httprequest.headers.get('api-key')
auth_api = self.auth_api_key(api_key)
model = kw.get('model')
username = request.httprequest.headers.get('login')
password = request.httprequest.headers.get('password')
credential = {'login': username, 'password': password, 'type': 'password'}
request.session.authenticate(request.session.db, credential)
model_id = request.env['ir.model'].search(
[('model', '=', model)])
if not model_id:
return ("<html><body><h3>Invalid model, check spelling or maybe "
"the related "
"module is not installed"
"</h3></body></html>")
if auth_api == True:
if not kw.get('Id'):
rec_id = 0
else: else:
rec_id = int(kw.get('Id')) # Obtener todos los registros
# Pass the query parameters for GET requests domain = []
result = self.generate_response(http_method, model_id.id, rec_id, kw) search_fields = fields if fields else ['id', 'display_name']
return result
else: records = model.search_read(domain=domain, fields=search_fields)
return auth_api serialized_records = self._serialize_record_values(records)
@http.route(['/odoo_connect'], type="http", auth="none", csrf=False, response_data = {
methods=['GET']) "success": True,
def odoo_connect(self, **kw): "count": len(serialized_records),
"""This is the controller which initializes the api transaction by "data": serialized_records
generating the api-key for specific user and database""" }
username = request.httprequest.headers.get('login')
password = request.httprequest.headers.get('password') return self._json_response(response_data)
db = request.httprequest.headers.get('db')
except Exception as e:
_logger.error(f"Error en GET: {str(e)}")
return self._error_response("Error obteniendo registros", 500)
def _handle_post(self, model, data, fields):
"""Maneja requests POST (crear)"""
if not data.get('values'):
return self._error_response("Se requiere 'values' para crear registro", 400)
try:
new_record = model.create(data['values'])
# Obtener el registro creado con los campos especificados
search_fields = fields if fields else ['id', 'display_name']
record_data = new_record.read(search_fields)[0]
serialized_record = self._serialize_record_values([record_data])
response_data = {
"success": True,
"message": "Registro creado exitosamente",
"data": serialized_record[0] if serialized_record else {}
}
return self._json_response(response_data, 201)
except Exception as e:
_logger.error(f"Error en POST: {str(e)}")
return self._error_response(f"Error creando registro: {str(e)}", 400)
def _handle_put(self, model, record_id, data, fields):
"""Maneja requests PUT (actualizar)"""
if not record_id:
return self._error_response("ID de registro requerido para actualización", 400)
if not data.get('values'):
return self._error_response("Se requiere 'values' para actualizar registro", 400)
try: try:
request.session.update(http.get_default_session(), db=db) record = model.browse(record_id)
credential = {'login': username, 'password': password, if not record.exists():
'type': 'password'} return self._error_response("Registro no encontrado", 404)
auth = request.session.authenticate(db, credential) record.write(data['values'])
user = request.env['res.users'].browse(auth['uid'])
api_key = request.env.user.generate_api(username) # Obtener el registro actualizado
datas = json.dumps({"Status": "auth successful", search_fields = fields if fields else ['id', 'display_name']
"User": user.name, record_data = record.read(search_fields)[0]
"api-key": api_key}) serialized_record = self._serialize_record_values([record_data])
return request.make_response(data=datas)
response_data = {
"success": True,
"message": "Registro actualizado exitosamente",
"data": serialized_record[0] if serialized_record else {}
}
return self._json_response(response_data)
except Exception as e:
_logger.error(f"Error en PUT: {str(e)}")
return self._error_response(f"Error actualizando registro: {str(e)}", 400)
def _handle_delete(self, model, record_id):
"""Maneja requests DELETE"""
if not record_id:
return self._error_response("ID de registro requerido para eliminación", 400)
try:
record = model.browse(record_id)
if not record.exists():
return self._error_response("Registro no encontrado", 404)
# Guardar información del registro antes de eliminarlo
record_info = {
"id": record.id,
"display_name": record.display_name if hasattr(record, 'display_name') else str(record)
}
record.unlink()
response_data = {
"success": True,
"message": "Registro eliminado exitosamente",
"deleted_record": record_info
}
return self._json_response(response_data)
except Exception as e:
_logger.error(f"Error en DELETE: {str(e)}")
return self._error_response(f"Error eliminando registro: {str(e)}", 400)
@http.route(['/api/v1/models'], type='http', auth='none', methods=['GET'], csrf=False)
def list_available_models(self, **kw):
"""Endpoint para listar modelos disponibles en la API"""
api_key = request.httprequest.headers.get('X-API-Key') or request.httprequest.headers.get('api-key')
success, user_id, error_msg = self._authenticate_api_key(api_key)
if not success:
return self._error_response(error_msg, 401)
try:
api_configs = request.env['connection.api'].sudo().search([])
models_data = []
for config in api_configs:
model_info = {
"model": config.model_id.model,
"name": config.model_id.name,
"methods": {
"GET": config.is_get,
"POST": config.is_post,
"PUT": config.is_put,
"DELETE": config.is_delete
}
}
models_data.append(model_info)
response_data = {
"success": True,
"count": len(models_data),
"data": models_data
}
return self._json_response(response_data)
except Exception as e: except Exception as e:
_logger.error(f"Error in authentication: {str(e)}") _logger.error(f"Error listando modelos: {str(e)}")
return ("<html><body><h2>wrong login credentials" return self._error_response("Error interno del servidor", 500)
"</h2></body></html>")

Write Preview
Loading…
Cancel
Save