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  # For decimal digits
    
    def encrypt(self, plaintext):
        """Encrypt using FF1"""
        n = len(plaintext)
        u = n // 2
        v = n - u
        
        A = plaintext[:u]
        B = plaintext[u:]
        
        # 10 rounds
        for i in range(10):
            # Calculate C
            C = self.calculate_C(A, i, u, v)
            
            # Calculate P
            P = self.calculate_P(C, i)
            
            # Calculate Y
            Y = self.calculate_Y(P)
            
            # Calculate Z
            Z = self.calculate_Z(Y, B)
            
            # Update A and 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):
        """Decrypt using FF1"""
        n = len(ciphertext)
        u = n // 2
        v = n - u
        
        A = ciphertext[:u]
        B = ciphertext[u:]
        
        # 10 rounds in reverse order
        for i in range(9, -1, -1):
            # Calculate C
            C = self.calculate_C(A, i, u, v)
            
            # Calculate P
            P = self.calculate_P(C, i)
            
            # Calculate Y
            Y = self.calculate_Y(P)
            
            # Calculate Z
            Z = self.calculate_Z(Y, B)
            
            # Update A and 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):
        """Calculate C according to FF1"""
        # Simplified implementation
        return f"{i:02d}{A}{u:02d}{v:02d}"
    
    def calculate_P(self, C, i):
        """Calculate P according to FF1"""
        # Simplified implementation
        return f"{C}{self.tweak.decode()}{i:02d}"
    
    def calculate_Y(self, P):
        """Calculate Y using AES"""
        cipher = Cipher(algorithms.AES(self.key), modes.ECB())
        encryptor = cipher.encryptor()
        
        # Padding for multiple of 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):
        """Calculate Z"""
        # Simplified implementation
        return Y[:len(B)]
    
    def add_mod_radix(self, a, b):
        """Addition modulo 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):
        """Subtraction modulo 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

# Usage example
key = b'1234567890123456'  # 16 bytes for AES-128
fpe = FF1Algorithm(key, tweak=b"tweak")

# Encrypt credit card number
credit_card = "4111111111111111"
encrypted = fpe.encrypt(credit_card)
print(f"Encrypted card: {encrypted}")

# Decrypt
decrypted = fpe.decrypt(encrypted)
print(f"Decrypted card: {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):
        """Encrypt credit card maintaining format"""
        # Remove spaces and dashes
        clean_card = credit_card.replace(' ', '').replace('-', '')
        
        # Verify format
        if not clean_card.isdigit() or len(clean_card) != 16:
            raise ValueError("Invalid card format")
        
        # Encrypt using FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_card)
        
        # Format as credit card
        formatted = f"{encrypted_digits[:4]}-{encrypted_digits[4:8]}-{encrypted_digits[8:12]}-{encrypted_digits[12:16]}"
        
        return formatted
    
    def encrypt_ssn(self, ssn):
        """Encrypt SSN maintaining format"""
        # Remove dashes
        clean_ssn = ssn.replace('-', '')
        
        # Verify format
        if not clean_ssn.isdigit() or len(clean_ssn) != 9:
            raise ValueError("Invalid SSN format")
        
        # Encrypt using FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_ssn)
        
        # Format as SSN
        formatted = f"{encrypted_digits[:3]}-{encrypted_digits[3:5]}-{encrypted_digits[5:9]}"
        
        return formatted
    
    def encrypt_email(self, email):
        """Encrypt email maintaining format"""
        # Separate username and domain
        if '@' not in email:
            raise ValueError("Invalid email format")
        
        username, domain = email.split('@', 1)
        
        # Encrypt username
        fpe = FF1Algorithm(self.key)
        encrypted_username = fpe.encrypt(username)
        
        # Keep domain
        formatted = f"{encrypted_username}@{domain}"
        
        return formatted

# Usage example
fpe_system = FormatPreservingEncryption(b'1234567890123456')

# Encrypt different formats
credit_card = "4111-1111-1111-1111"
encrypted_card = fpe_system.encrypt_credit_card(credit_card)
print(f"Encrypted card: {encrypted_card}")

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

email = "john.doe@example.com"
encrypted_email = fpe_system.encrypt_email(email)
print(f"Encrypted email: {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):
        """Protect database field"""
        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):
        """Encrypt phone number"""
        # Remove non-numeric characters
        clean_phone = ''.join(filter(str.isdigit, phone))
        
        # Verify format
        if len(clean_phone) != 10:
            raise ValueError("Invalid phone format")
        
        # Encrypt using FF1
        fpe = FF1Algorithm(self.key)
        encrypted_digits = fpe.encrypt(clean_phone)
        
        # Format as phone
        formatted = f"({encrypted_digits[:3]}) {encrypted_digits[3:6]}-{encrypted_digits[6:10]}"
        
        return formatted

# Usage example
data_protection = SensitiveDataProtection(b'1234567890123456')

# Protect different data types
credit_card = "4111-1111-1111-1111"
protected_card = data_protection.protect_database_field(credit_card, 'credit_card')
print(f"Protected card: {protected_card}")

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

phone = "(555) 123-4567"
protected_phone = data_protection.protect_database_field(phone, 'phone')
print(f"Protected phone: {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):
        """Migrate database applying FPE"""
        # Get sensitive fields
        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):
        """Get sensitive fields from database"""
        # Implement logic to identify sensitive fields
        return {
            'users': ['ssn', 'credit_card'],
            'orders': ['payment_info'],
            'customers': ['phone', 'email']
        }
    
    def encrypt_field(self, db_conn, table, field):
        """Encrypt specific field"""
        # Get field data
        cursor = db_conn.cursor()
        cursor.execute(f"SELECT id, {field} FROM {table}")
        rows = cursor.fetchall()
        
        # Encrypt each value
        for row_id, value in rows:
            if value:
                encrypted_value = self.fpe.encrypt_credit_card(value)  # Simplified
                cursor.execute(
                    f"UPDATE {table} SET {field} = %s WHERE id = %s",
                    (encrypted_value, row_id)
                )
        
        db_conn.commit()

# Usage example
# migration = DataMigration(b'1234567890123456')
# migration.migrate_database(database_connection)