nahakubuilder/PyMTA-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dkim test

Browse Source
This commit is contained in:
nahakubuilde
2025-05-31 17:35:35 +01:00
parent 1d4a37922d
commit 06f75d704d
3 changed files with 138 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
email_server/dkim_manager.py
View File

@@ -9,6 +9,8 @@ from datetime import datetime
from email_server.models import Session, Domain, DKIMKey
from email_server.settings_loader import load_settings
from email_server.tool_box import get_logger
import random
import string
settings = load_settings()
DKIM_SELECTOR = settings['DKIM']['DKIM_SELECTOR']
@@ -19,11 +21,20 @@ logger = get_logger()
class DKIMManager:
"""Manages DKIM keys and email signing."""
def __init__(self):
self.selector = DKIM_SELECTOR
def __init__(self, selector: str = None):
"""Initialize DKIMManager with a selector. If not provided, use random."""
if selector:
self.selector = selector
else:
self.selector = self._generate_random_selector()
def generate_dkim_keypair(self, domain_name):
"""Generate DKIM key pair for a domain."""
@staticmethod
def _generate_random_selector(length: int = 12) -> str:
"""Generate a random DKIM selector name (8-12 chars)."""
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
def generate_dkim_keypair(self, domain_name, selector: str = None):
"""Generate DKIM key pair for a domain, optionally with a custom selector."""
session = Session()
try:
# Check if domain exists
@@ -32,10 +43,13 @@ class DKIMManager:
logger.error(f"Domain {domain_name} not found")
return False
# Check if DKIM key already exists
existing_key = session.query(DKIMKey).filter_by(domain_id=domain.id, is_active=True).first()
# Use provided selector or instance selector
use_selector = selector or self.selector
# Check if DKIM key with this selector already exists
existing_key = session.query(DKIMKey).filter_by(domain_id=domain.id, selector=use_selector, is_active=True).first()
if existing_key:
logger.debug(f"DKIM key already exists for domain {domain_name}")
logger.debug(f"DKIM key already exists for domain {domain_name} and selector {use_selector}")
return True
# Generate RSA key pair
@@ -61,7 +75,7 @@ class DKIMManager:
# Store in database
dkim_key = DKIMKey(
domain_id=domain.id,
selector=self.selector,
selector=use_selector,
private_key=private_pem,
public_key=public_pem,
created_at=datetime.now(),
@@ -70,7 +84,7 @@ class DKIMManager:
session.add(dkim_key)
session.commit()
logger.debug(f"Generated DKIM key for domain: {domain_name}")
logger.debug(f"Generated DKIM key for domain: {domain_name} selector: {use_selector}")
return True
except Exception as e:
@@ -79,95 +93,78 @@ class DKIMManager:
return False
finally:
session.close()
def get_active_dkim_key(self, domain_name):
"""Get the active DKIM key for a domain (only one active per selector)."""
session = Session()
try:
domain = session.query(Domain).filter_by(domain_name=domain_name).first()
if not domain:
return None
dkim_key = session.query(DKIMKey).filter_by(domain_id=domain.id, is_active=True).first()
return dkim_key
except Exception as e:
logger.error(f"Error getting DKIM key for {domain_name}: {e}")
return None
finally:
session.close()
def get_dkim_private_key(self, domain_name):
"""Get DKIM private key for a domain."""
session = Session()
try:
domain = session.query(Domain).filter_by(domain_name=domain_name).first()
if not domain:
return None
dkim_key = session.query(DKIMKey).filter_by(
domain_id=domain.id,
is_active=True
).first()
if dkim_key:
return dkim_key.private_key
return None
except Exception as e:
logger.error(f"Error getting DKIM private key for {domain_name}: {e}")
return None
finally:
session.close()
dkim_key = self.get_active_dkim_key(domain_name)
if dkim_key:
return dkim_key.private_key
return None
def get_dkim_public_key_record(self, domain_name):
"""Get DKIM public key DNS record for a domain."""
session = Session()
try:
domain = session.query(Domain).filter_by(domain_name=domain_name).first()
if not domain:
return None
dkim_key = session.query(DKIMKey).filter_by(
domain_id=domain.id,
is_active=True
).first()
if dkim_key:
# Extract public key from PEM format for DNS record
public_key_lines = dkim_key.public_key.strip().split('\n')
public_key_data = ''.join(public_key_lines[1:-1]) # Remove header/footer
return {
'name': f'{self.selector}._domainkey.{domain_name}',
'type': 'TXT',
'value': f'v=DKIM1; k=rsa; p={public_key_data}'
}
return None
except Exception as e:
logger.error(f"Error getting DKIM public key record for {domain_name}: {e}")
return None
finally:
session.close()
"""Get DKIM public key DNS record for a domain (active key only)."""
dkim_key = self.get_active_dkim_key(domain_name)
if dkim_key:
public_key_lines = dkim_key.public_key.strip().split('\n')
public_key_data = ''.join(public_key_lines[1:-1]) # Remove header/footer
return {
'name': f'{dkim_key.selector}._domainkey.{domain_name}',
'type': 'TXT',
'value': f'v=DKIM1; k=rsa; p={public_key_data}'
}
return None
def sign_email(self, email_content, domain_name):
"""Sign email content with DKIM."""
"""Sign email content with DKIM. Only add one DKIM header, after all modifications."""
try:
private_key = self.get_dkim_private_key(domain_name)
if not private_key:
dkim_key = self.get_active_dkim_key(domain_name)
if not dkim_key:
logger.warning(f"No DKIM key found for domain: {domain_name}")
return email_content
private_key = dkim_key.private_key
selector = dkim_key.selector
# Convert content to bytes if it's a string
if isinstance(email_content, str):
email_bytes = email_content.encode('utf-8')
else:
email_bytes = email_content
# Sign the email
# Remove any existing DKIM-Signature header (robust multiline)
import re
email_bytes = re.sub(br'^DKIM-Signature:.*?(\r?\n[ \t].*?)*\r?\n', b'', email_bytes, flags=re.MULTILINE)
# Canonicalization: relaxed/relaxed, add more headers to h=
headers = [
b'from', b'to', b'subject', b'date', b'message-id',
b'mime-version', b'content-type', b'content-transfer-encoding'
]
signature = dkim.sign(
email_bytes,
self.selector.encode('utf-8'),
selector.encode('utf-8'),
domain_name.encode('utf-8'),
private_key.encode('utf-8'),
include_headers=[b'from', b'to', b'subject', b'date', b'message-id']
include_headers=headers,
canonicalize=(b'relaxed', b'relaxed')
)
# Combine signature with original content
signed_content = signature + email_bytes
logger.debug(f"Successfully signed email for domain: {domain_name}")
# Return as string if input was string
logger.debug(f"Successfully signed email for domain: {domain_name} selector: {selector}")
if isinstance(email_content, str):
return signed_content.decode('utf-8')
else:
return signed_content
except Exception as e:
logger.error(f"Error signing email for domain {domain_name}: {e}")
return email_content

5
email_server/smtp_handler.py
View File

@@ -61,17 +61,16 @@ class CustomSMTPHandler:
# Extract domain from sender for DKIM signing
sender_domain = envelope.mail_from.split('@')[1] if '@' in envelope.mail_from else None
# Sign with DKIM if domain is configured
# Relay the email (all modifications done)
signed_content = content
dkim_signed = False
if sender_domain:
# DKIM-sign the final version of the message
signed_content = self.dkim_manager.sign_email(content, sender_domain)
# Check if signing was successful (content changed)
dkim_signed = signed_content != content
if dkim_signed:
logger.debug(f'Email {message_id} signed with DKIM for domain {sender_domain}')
# Relay the email
success = self.email_relay.relay_email(
envelope.mail_from,
envelope.rcpt_tos,