FEAT: Implement email encryption and decryption in community model; update related functions and tests

Author: IkeSalmonson

app/routers/admin/routes.py

@@ -61,7 +61,8 @@ async def post_create_community(
         admin_role = admin_community.role
         if admin_role != "admin":
             return {"status": "Unauthorized"}
-        await create_community(request=request, community=community)
+        session: AsyncSession = request.app.db_session_factory
+        await create_community(session=session, community=community)
 
         return CommunityPostResponse()
 

app/services/database/models/communities.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from typing import Optional
 
+from sqlalchemy import Text
 from sqlmodel import Field, SQLModel
 
 
@@ -9,9 +10,8 @@ class Community(SQLModel, table=True):
 
     id: Optional[int] = Field(default=None, primary_key=True)
     username: str
-    email: str
+    email: str = Field(sa_column=Text)  # VARCHAR(255)
     password: str
-    role: str = Field(default="user")
     created_at: Optional[datetime] = Field(default_factory=datetime.now)
     updated_at: Optional[datetime] = Field(
         default_factory=datetime.now,

app/services/database/orm/community.py

@@ -1,10 +1,10 @@
 from typing import Optional
 
-from fastapi import Request
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.services.database.models import Community
+from app.services.encryption import decrypt_email, encrypt_email
 
 
 async def get_community_by_username(
@@ -21,19 +21,22 @@ async def get_community_by_username(
     # Executa a declaração na sessão e retorna o primeiro resultado
     result = await session.exec(statement)
     community = result.first()
+    # add tratamento de descriptografia do email
+    if community is not None:
+        community.email = decrypt_email(community.email)
 
     return community
 
 
 async def create_community(
-    request: Request,
+    session: AsyncSession,
     community: Community,  # community model
 ) -> Optional[Community]:
     """
     Cria um novo membro da comunidade.
     Somente usuário autenticado e com role Admin podem executar.
     """
-    session: AsyncSession = request.app.db_session_factory
+    community.email = encrypt_email(community.email)
     session.add(community)
     await session.commit()
     await session.refresh(community)

app/services/encryption.py

@@ -3,6 +3,7 @@
 from cryptography.fernet import Fernet
 
 test_encryption_key = "r0-QKv5qACJNFRqy2cNZCsfZ_zVvehlC-v8zDJb--EI="
+# print(Fernet.generate_key())
 # Carrega a chave da variável de ambiente
 ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY", test_encryption_key)
 
@@ -11,14 +12,19 @@
         "ENCRYPTION_KEY não está definida nas variáveis de ambiente."
     )
 
-cipher = Fernet(ENCRYPTION_KEY.encode())
+cipher = Fernet(ENCRYPTION_KEY)
 
 
 def encrypt_email(email: str) -> str:
     """Criptografa uma string de e-mail."""
-    return cipher.encrypt(email.encode()).decode()
+    #    var = cipher.encrypt(b'{email}')
+
+    var = cipher.encrypt(email.encode("utf-8"))
+    print(f"encrypt_email: {email} -> {var}")
+    print(len(var))
+    return var
 
 
 def decrypt_email(encrypted_email: str) -> str:
     """Descriptografa uma string de e-mail."""
-    return cipher.decrypt(encrypted_email.encode()).decode()
+    return cipher.decrypt(encrypted_email).decode("utf-8")

docker-compose.yaml

@@ -14,7 +14,7 @@ services:
       - SQLITE_PATH=app/services/database/pynewsdb.db
       - SQLITE_URL=sqlite+aiosqlite://
       - SECRET_KEY=1a6c5f3b7d2e4a7fb68d0casd3f9a7b2d8c4e5f6a3b0d4e9c7a8f1b6d3c0a7f5e
-      - ENCRYPTION_KEY=smR739opNB9FJ4hEm5ZIG8Gr-Qnvqtem4ehwl4RIUes=
+      - ENCRYPTION_KEY=r0-QKv5qACJNFRqy2cNZCsfZ_zVvehlC-v8zDJb--EI=
       - ADMIN_USER=admin
       - ADMIN_PASSWORD=admin
       - ALGORITHM=HS256

tests/test_communities.py

@@ -3,24 +3,104 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.services.database.models import Community
+from app.services.database.orm.community import (
+    create_community,
+    get_community_by_username,
+)
+
+# Dados de teste
+TEST_USERNAME = "test_user_crypto"
+TEST_EMAIL = "crypto@test.com"
+TEST_PASSWORD = "@SafePassword123"
 
 
 @pytest.mark.asyncio
 async def test_insert_communities(session: AsyncSession):
     community = Community(
-        username="admin",
-        email="teste@teste.com",
-        password="@teste123",
+        username=TEST_USERNAME,
+        email=TEST_EMAIL,
+        password=TEST_PASSWORD,
     )
     session.add(community)
     await session.commit()
     await session.refresh(community)
 
-    statement = select(Community).where(Community.username == "admin")
+    statement = select(Community).where(Community.username == TEST_USERNAME)
     result = await session.exec(statement)
     found = result.first()
 
     assert found is not None
-    assert found.username == "admin"
-    assert found.email == "teste@teste.com"
-    assert found.password == "@teste123"
+    assert found.username == TEST_USERNAME
+    assert found.email == TEST_EMAIL
+    assert found.password == TEST_PASSWORD
+
+
+@pytest.mark.asyncio
+async def test_community_orm_flow_with_encryption_transparency(
+    session: AsyncSession,
+):
+    """
+    Testa a criação e a leitura de uma comunidade, validando
+    que o ORM (propriedades) garante a transparência da criptografia.
+    """
+    new_community = Community(
+        username=TEST_USERNAME,
+        email=TEST_EMAIL,
+        password=TEST_PASSWORD,
+    )
+
+    # 2. Ação: Use a função ORM para criar a comunidade
+    created_community = await create_community(
+        community=new_community, session=session
+    )
+
+    # 3. Asserção de Leitura Transparente (getters)
+    # Ao acessar 'created_community.email', ele DEVE retornar o email descriptografado
+    assert created_community.email == TEST_EMAIL
+    assert created_community.username == TEST_USERNAME
+
+    # 4. Asserção da Criptografia (Validação do Armazenamento)
+    # Acessar o campo interno '_email' para provar que está criptografado
+    stored_email = created_community._email
+
+    # O email armazenado não deve ser igual ao email original (em texto puro)
+    assert stored_email != TEST_EMAIL
+
+    # O email armazenado deve ser um valor válido de Fernet (a criptografia)
+    # Usamos a função de criptografia para ter um valor esperado
+    assert stored_email == TEST_EMAIL
+
+
+@pytest.mark.asyncio
+async def test_get_community_by_username_orm(session: AsyncSession):
+    """
+    Testa a função de leitura 'get_community_by_username' e a descriptografia.
+    """
+    # 1. Preparação: Crie um registro diretamente no banco para garantir
+    # que o teste não dependa da função create_community
+    community_to_insert = Community(
+        username="newreader_test",
+        email=TEST_EMAIL,
+        password=TEST_PASSWORD,
+    )
+
+    # Fazemos a inserção no banco de forma manual para forçar a criptografia
+    # (O setter do modelo faz a criptografia automaticamente aqui)
+    session.add(community_to_insert)
+    await session.commit()
+    await session.refresh(community_to_insert)
+
+    # 2. Ação: Use a função ORM para buscar o registro
+    found_community = await get_community_by_username(
+        username="newreader_test", session=session
+    )
+
+    # 3. Asserções
+    assert found_community is not None
+
+    # O email lido deve ser o valor original (descriptografado pelo modelo)
+    assert found_community.email == TEST_EMAIL
+    assert found_community.username == "newreader_test"
+
+    # Garante que a senha está correta (embora não seja o foco, é bom manter)
+    assert found_community.password == TEST_PASSWORD