-
-
Save rdhyee/05df014d9d67f304906c97fda6ffc1c1 to your computer and use it in GitHub Desktop.
Implementation of the AES protocol as used by https://encipher.it
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
# -*- coding: utf-8 -*- | |
# | |
# implementation of the AES protocol as used by https://encipher.it | |
# | |
# (c) 2018 Bernd Busse | |
from cryptography.hazmat.primitives.ciphers import Cipher | |
from cryptography.hazmat.primitives.ciphers.algorithms import AES | |
from cryptography.hazmat.primitives.ciphers.modes import ECB, CTR | |
from cryptography.hazmat.primitives.hashes import SHA1 | |
from cryptography.hazmat.primitives.hmac import HMAC | |
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC | |
from cryptography.exceptions import InvalidSignature | |
from cryptography.hazmat.backends import default_backend | |
from datetime import datetime | |
import base64 | |
import binascii | |
import secrets | |
import sys | |
import struct | |
import time | |
ENCT2_HEAD = "EnCt2" | |
ENCT2_FOOT = "IwEmS" | |
class EncryptionError(RuntimeError): | |
pass | |
class DecryptionError(RuntimeError): | |
pass | |
def usage(executable): | |
sys.stderr.write("usage: {} -e KEY TEXT\n".format(executable)) | |
sys.stderr.write(" {} -d KEY CIPHER\n".format(executable)) | |
sys.stderr.write(" {} -a CIPHER\n".format(executable)) | |
sys.stderr.flush() | |
def encrypt(password, message): | |
# Derive key (using PBKDF2) | |
mac_salt = base64.b64encode(secrets.token_bytes(6)) | |
mac_kdf = PBKDF2HMAC(algorithm=SHA1(), | |
length=32, | |
salt=mac_salt, | |
iterations=1000, | |
backend=default_backend()) | |
mac_key = mac_kdf.derive(password.encode("utf-8")) | |
mac_key = binascii.hexlify(mac_key) | |
# Calculate HMAC over the message (using SHA1) | |
mac = HMAC(key=mac_key, | |
algorithm=SHA1(), | |
backend=default_backend()) | |
mac.update(message.encode("utf-8")) | |
c_hmac = mac.finalize() | |
c_hmac = binascii.hexlify(c_hmac) | |
# Generate nonce for counter (using timestamp) | |
ts = int(time.time()*1000) | |
ctr_nonce = ((ts % 1000) & 0xffff).to_bytes(2, "little") | |
ctr_nonce += secrets.token_bytes(2) | |
ctr_nonce += ((ts//1000) & 0xffffffff).to_bytes(4, "little") | |
# Generate new key for encryption (using AES) | |
keycipher = Cipher(algorithm=AES(mac_key[:32]), | |
mode=ECB(), | |
backend=default_backend()) | |
keycipher = keycipher.encryptor() | |
aes_key = keycipher.update(mac_key[:16]) + keycipher.finalize() | |
aes_key += aes_key | |
# Encrypt the message (using AES in CTR-mode) | |
cipher = Cipher(algorithm=AES(aes_key), | |
mode=CTR(ctr_nonce + bytes(8)), | |
backend=default_backend()) | |
cipher = cipher.encryptor() | |
c_text = cipher.update(message.encode("utf-8")) + cipher.finalize() | |
return ENCT2_HEAD + \ | |
c_hmac.decode("utf-8") + c_hmac[:24].decode("utf-8") + \ | |
mac_salt.decode("utf-8") + \ | |
base64.b64encode(ctr_nonce + c_text).decode("utf-8") + \ | |
ENCT2_FOOT | |
def analyze(ciphertext): | |
# Look for header and footer | |
if ciphertext[:5] != ENCT2_HEAD or ciphertext[-5:] != ENCT2_FOOT: | |
raise DecryptionError("Invalid cipher format") | |
ct = ciphertext[5:-5] # trim head and foot | |
# Extract parts | |
c_hmac = ct[:40] | |
mac_salt = ct[64:72].encode('utf8') | |
c_text = base64.b64decode(ct[72:]) | |
ctr_nonce, c_text = c_text[:8], c_text[8:] | |
return (c_hmac, mac_salt, ctr_nonce, c_text) | |
def decrypt(password, ciphertext): | |
# Extract parts | |
c_hmac, mac_salt, ctr_nonce, c_text = analyze(ciphertext) | |
# Derive key (using PBKDF2) | |
mac_kdf = PBKDF2HMAC(algorithm=SHA1(), | |
length=32, | |
salt=mac_salt, | |
iterations=1000, | |
backend=default_backend()) | |
mac_key = mac_kdf.derive(password.encode("utf-8")) | |
mac_key = binascii.hexlify(mac_key) | |
# Generate key for decryption (using AES) | |
keycipher = Cipher(algorithm=AES(mac_key[:32]), | |
mode=ECB(), | |
backend=default_backend()) | |
keycipher = keycipher.encryptor() | |
aes_key = keycipher.update(mac_key[:16]) + keycipher.finalize() | |
aes_key += aes_key | |
# Decrypt the ciphertext (using AES in CTR-mode) | |
cipher = Cipher(algorithm=AES(aes_key), | |
mode=CTR(ctr_nonce + bytes(8)), | |
backend=default_backend()) | |
cipher = cipher.decryptor() | |
message = cipher.update(c_text) + cipher.finalize() | |
# Check the HMAC over the message (using SHA1) | |
mac = HMAC(key=mac_key, | |
algorithm=SHA1(), | |
backend=default_backend()) | |
mac.update(message) | |
try: | |
mac.verify(binascii.unhexlify(c_hmac)) | |
except InvalidSignature as e: | |
raise DecryptionError("Invalid HMAC") from e | |
return message.decode("utf-8") | |
if __name__ == "__main__": | |
if len(sys.argv) < 3 or len(sys.argv) > 5: | |
usage(sys.argv[0]) | |
exit(1) | |
if sys.argv[1] == "-e": | |
# Run encryption | |
cipher = encrypt(sys.argv[2], sys.argv[3]) | |
print("Ciphertext: {:s}".format(cipher)) | |
elif sys.argv[1] == "-d": | |
# Run decryption | |
plain = decrypt(sys.argv[2], sys.argv[3]) | |
print("Plaintext: {:s}".format(plain)) | |
elif sys.argv[1] == "-a": | |
# Analyze ciphertext | |
hmac, salt, nonce, text = analyze(sys.argv[2]) | |
ms, _, sec = struct.unpack("<HHI", nonce) | |
ts = (sec * 1000 + ms) / 1000 | |
date = datetime.utcfromtimestamp(ts) | |
print("Message Structure:") | |
print(" HMAC:\t\t{:s}\n Salt:\t\t{:s}" | |
.format(hmac, salt.decode('utf8'))) | |
print(" Nonce:\t{:s}\n Ciphertext:\t{:s}" | |
.format(binascii.hexlify(nonce).decode('utf8'), | |
binascii.hexlify(text).decode('utf8'))) | |
print(" Created at:\t{:d} ({:s})" | |
.format(int(ts), date.isoformat())) | |
else: | |
usage(sys.argv[0]) | |
exit(1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment