106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import Column, Integer, String
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from backend.ConexionSql.Base_conexion import ConexionBase
|
|
from backend.base import Base
|
|
from backend.Credenciales.postgres_credencial import PostgresCredencial
|
|
from security.Encriptar import Encriptar_fernet
|
|
|
|
# ----------------------
|
|
# Cargar clave maestra
|
|
# ----------------------
|
|
from entrypoint import ENV_PATH
|
|
load_dotenv(ENV_PATH)
|
|
pssword = os.getenv('MASTER_PASSWORD')
|
|
if pssword is None:
|
|
raise ValueError("MASTER_PASSWORD no está definida en el archivo .env")
|
|
|
|
# ----------------------
|
|
# MODELO (SQLAlchemy)
|
|
# ----------------------
|
|
|
|
class PostgresCredencialModel(Base):
|
|
__tablename__ = 'postgres_credenciales'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
titulo = Column(String, nullable=False)
|
|
host = Column(String, nullable=False)
|
|
port = Column(Integer, nullable=False)
|
|
dbname = Column(String, nullable=False)
|
|
user = Column(String, nullable=False)
|
|
password = Column(String, nullable=False) # Encriptada como base64 string
|
|
|
|
# ----------------------
|
|
# MAPPER
|
|
# ----------------------
|
|
|
|
import base64
|
|
|
|
class PostgresCredencialMapper:
|
|
@staticmethod
|
|
def to_dict(obj: PostgresCredencial) -> dict:
|
|
return {
|
|
"titulo": obj.titulo,
|
|
"host": obj.host,
|
|
"port": obj.port,
|
|
"dbname": obj.dbname,
|
|
"user": obj.user,
|
|
"password": base64.b64encode(
|
|
Encriptar_fernet.encriptar(obj.password, pssword)
|
|
).decode('utf-8')
|
|
}
|
|
|
|
@staticmethod
|
|
def from_dict(data: dict) -> PostgresCredencial:
|
|
return PostgresCredencial(
|
|
titulo=data["titulo"],
|
|
host=data["host"],
|
|
port=data["port"],
|
|
dbname=data["dbname"],
|
|
user=data["user"],
|
|
password=Encriptar_fernet.desencriptar(
|
|
base64.b64decode(data["password"]), pssword
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def from_model(model: PostgresCredencialModel) -> PostgresCredencial:
|
|
return PostgresCredencial(
|
|
titulo=model.titulo,
|
|
host=model.host,
|
|
port=model.port,
|
|
dbname=model.dbname,
|
|
user=model.user,
|
|
password=Encriptar_fernet.desencriptar(
|
|
base64.b64decode(model.password), pssword
|
|
)
|
|
)
|
|
|
|
# ----------------------
|
|
# REPO
|
|
# ----------------------
|
|
|
|
class PostgresCredencialRepo:
|
|
def __init__(self, conexion: ConexionBase):
|
|
self.session = conexion.get_session()
|
|
|
|
def add(self, credencial: PostgresCredencial) -> int:
|
|
data = PostgresCredencialMapper.to_dict(credencial)
|
|
model = PostgresCredencialModel(**data)
|
|
self.session.add(model)
|
|
self.session.commit()
|
|
return model.id
|
|
|
|
def get_all(self) -> list[PostgresCredencial]:
|
|
models = self.session.query(PostgresCredencialModel).all()
|
|
return [PostgresCredencialMapper.from_model(m) for m in models]
|
|
|
|
def get_by_titulo(self, titulo: str) -> PostgresCredencial | None:
|
|
model = self.session.query(PostgresCredencialModel).filter_by(titulo=titulo).first()
|
|
return PostgresCredencialMapper.from_model(model) if model else None
|
|
|
|
def get_by_id(self, id_: int) -> PostgresCredencial | None:
|
|
model = self.session.get(PostgresCredencialModel, id_)
|
|
return PostgresCredencialMapper.from_model(model) if model else None |