Format-Preserving Encryption

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
import math

class FF1Algorithm:
    def __init__(self, key, tweak=b""):
        self.key = key
        self.tweak = tweak
        self.radix = 10  # Para dígitos decimales
    
    def encrypt(self, plaintext):
        """Cifrar usando FF1"""
        n = len(plaintext)
        u = n // 2
        v = n - u
        
        A = plaintext[:u]
        B = plaintext[u:]
        
        # 10 rondas
        for i in range(10):
            # Calcular C
            C = self.calculate_C(A, i, u, v)
            
            # Calcular P
            P = self.calculate_P(C, i)
            
            # Calcular Y
            Y = self.calculate_Y(P)
            
            # Calcular Z
            Z = self.calculate_Z(Y, B)
            
            # Actualizar A y B
            A_new = B
            B_new = self.subtract_mod_radix(A, Z)
            
            A = A_new
            B = B_new
        
        return A + B
    
    def decrypt(self, ciphertext):
        """Descifrar usando FF1"""
        n = len(ciphertext)
        u = n // 2
        v = n - u
        
        A = ciphertext[:u]
        B = ciphertext[u:]
        
        # 10 rondas en orden inverso
        for i in range(9, -1, -1):
            # Calcular C
            C = self.calculate_C(A, i, u, v)
            
            # Calcular P
            P = self.calculate_P(C, i)
            
            # Calcular Y
            Y = self.calculate_Y(P)
            
            # Calcular Z
            Z = self.calculate_Z(Y, B)
            
            # Actualizar A y B
            B_new = A
            A_new = self.add_mod_radix(B, Z)
            
            A = A_new
            B = B_new
        
        return A + B
    
    def calculate_C(self, A, i, u, v):
        """Calcular C según FF1"""
        # Implementación simplificada
        return f"{i:02d}{A}{u:02d}{v:02d}"
    
    def calculate_P(self, C, i):
        """Calcular P según FF1"""
        # Implementación simplificada
        return f"{C}{self.tweak.decode()}{i:02d}"
    
    def calculate_Y(self, P):
        """Calcular Y usando AES"""
        cipher = Cipher(algorithms.AES(self.key), modes.ECB())
        encryptor = cipher.encryptor()
        
        # Padding para múltiplo de 16 bytes
        padded_P = P.encode().ljust(16, b'\x00')
        encrypted = encryptor.update(padded_P) + encryptor.finalize()
        
        return encrypted
    
    def calculate_Z(self, Y, B):
        """Calcular Z"""
        # Implementación simplificada
        return Y[:len(B)]
    
    def add_mod_radix(self, a, b):
        """Suma módulo radix"""
        result = ""
        carry = 0
        
        for i in range(len(a) - 1, -1, -1):
            sum_digits = int(a[i]) + int(b[i]) + carry
            result = str(sum_digits % self.radix) + result
            carry = sum_digits // self.radix
        
        return result
    
    def subtract_mod_radix(self, a, b):
        """Resta módulo radix"""
        result = ""
        borrow = 0
        
        for i in range(len(a) - 1, -1, -1):
            diff = int(a[i]) - int(b[i]) - borrow
            if diff < 0:
                diff += self.radix
                borrow = 1
            else:
                borrow = 0
            result = str(diff) + result
        
        return result

# Ejemplo de uso
key = b'1234567890123456'  # 16 bytes para AES-128
fpe = FF1Algorithm(key, tweak=b"tweak")

# Cifrar número de tarjeta de crédito
credit_card = "4111111111111111"
encrypted = fpe.encrypt(credit_card)
print(f"Tarjeta cifrada: {encrypted}")

# Descifrar
decrypted = fpe.decrypt(encrypted)
print(f"Tarjeta descifrada: {decrypted}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

class FormatPreservingEncryption:
    def __init__(self, key):
        self.key = key
    
    def encrypt_credit_card(self, credit_card):
        """Cifrar tarjeta de crédito manteniendo formato"""
        # Remover espacios y guiones
        clean_card = credit_card.replace(' ', '').replace('-', '')
        
        # Verificar formato
        if not clean_card.isdigit() or len(clean_card) != 16:
            raise ValueError("Formato de tarjeta inválido")
        
        # Cifrar usando FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_card)
        
        # Formatear como tarjeta de crédito
        formatted = f"{encrypted_digits[:4]}-{encrypted_digits[4:8]}-{encrypted_digits[8:12]}-{encrypted_digits[12:16]}"
        
        return formatted
    
    def encrypt_ssn(self, ssn):
        """Cifrar SSN manteniendo formato"""
        # Remover guiones
        clean_ssn = ssn.replace('-', '')
        
        # Verificar formato
        if not clean_ssn.isdigit() or len(clean_ssn) != 9:
            raise ValueError("Formato de SSN inválido")
        
        # Cifrar usando FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_ssn)
        
        # Formatear como SSN
        formatted = f"{encrypted_digits[:3]}-{encrypted_digits[3:5]}-{encrypted_digits[5:9]}"
        
        return formatted
    
    def encrypt_email(self, email):
        """Cifrar email manteniendo formato"""
        # Separar usuario y dominio
        if '@' not in email:
            raise ValueError("Formato de email inválido")
        
        username, domain = email.split('@', 1)
        
        # Cifrar username
        fpe = FF1Algorithm(self.key)
        encrypted_username = fpe.encrypt(username)
        
        # Mantener dominio
        formatted = f"{encrypted_username}@{domain}"
        
        return formatted

# Ejemplo de uso
fpe_system = FormatPreservingEncryption(b'1234567890123456')

# Cifrar diferentes formatos
credit_card = "4111-1111-1111-1111"
encrypted_card = fpe_system.encrypt_credit_card(credit_card)
print(f"Tarjeta cifrada: {encrypted_card}")

ssn = "123-45-6789"
encrypted_ssn = fpe_system.encrypt_ssn(ssn)
print(f"SSN cifrado: {encrypted_ssn}")

email = "john.doe@example.com"
encrypted_email = fpe_system.encrypt_email(email)
print(f"Email cifrado: {encrypted_email}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

class SensitiveDataProtection:
    def __init__(self, key):
        self.key = key
        self.fpe = FormatPreservingEncryption(key)
    
    def protect_database_field(self, data, field_type):
        """Proteger campo de base de datos"""
        if field_type == 'credit_card':
            return self.fpe.encrypt_credit_card(data)
        elif field_type == 'ssn':
            return self.fpe.encrypt_ssn(data)
        elif field_type == 'phone':
            return self.fpe.encrypt_phone(data)
        else:
            return data
    
    def encrypt_phone(self, phone):
        """Cifrar número de teléfono"""
        # Remover caracteres no numéricos
        clean_phone = ''.join(filter(str.isdigit, phone))
        
        # Verificar formato
        if len(clean_phone) != 10:
            raise ValueError("Formato de teléfono inválido")
        
        # Cifrar usando FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_phone)
        
        # Formatear como teléfono
        formatted = f"({encrypted_digits[:3]}) {encrypted_digits[3:6]}-{encrypted_digits[6:10]}"
        
        return formatted

# Ejemplo de uso
data_protection = SensitiveDataProtection(b'1234567890123456')

# Proteger diferentes tipos de datos
credit_card = "4111-1111-1111-1111"
protected_card = data_protection.protect_database_field(credit_card, 'credit_card')
print(f"Tarjeta protegida: {protected_card}")

ssn = "123-45-6789"
protected_ssn = data_protection.protect_database_field(ssn, 'ssn')
print(f"SSN protegido: {protected_ssn}")

phone = "(555) 123-4567"
protected_phone = data_protection.protect_database_field(phone, 'phone')
print(f"Teléfono protegido: {protected_phone}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

class DataMigration:
    def __init__(self, key):
        self.key = key
        self.fpe = FormatPreservingEncryption(key)
    
    def migrate_database(self, database_connection):
        """Migrar base de datos aplicando FPE"""
        # Obtener campos sensibles
        sensitive_fields = self.get_sensitive_fields(database_connection)
        
        for table, fields in sensitive_fields.items():
            for field in fields:
                self.encrypt_field(database_connection, table, field)
    
    def get_sensitive_fields(self, db_conn):
        """Obtener campos sensibles de la base de datos"""
        # Implementar lógica para identificar campos sensibles
        return {
            'users': ['ssn', 'credit_card'],
            'orders': ['payment_info'],
            'customers': ['phone', 'email']
        }
    
    def encrypt_field(self, db_conn, table, field):
        """Cifrar campo específico"""
        # Obtener datos del campo
        cursor = db_conn.cursor()
        cursor.execute(f"SELECT id, {field} FROM {table}")
        rows = cursor.fetchall()
        
        # Cifrar cada valor
        for row_id, value in rows:
            if value:
                encrypted_value = self.fpe.encrypt_credit_card(value)  # Simplificado
                cursor.execute(
                    f"UPDATE {table} SET {field} = %s WHERE id = %s",
                    (encrypted_value, row_id)
                )
        
        db_conn.commit()

# Ejemplo de uso
# migration = DataMigration(b'1234567890123456')
# migration.migrate_database(database_connection)