235 lines
6.9 KiB
Python
Executable File
235 lines
6.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
IP DIRECT SENDER
|
|
Méthode économique: IP + PTR seulement
|
|
Sans O365, Gmail, AWS SES, etc.
|
|
|
|
Utilise:
|
|
- PTR du serveur comme domaine sender
|
|
- SPF hérité ou custom
|
|
- PMTA pour delivery
|
|
"""
|
|
|
|
import psycopg2
|
|
import smtplib
|
|
import hashlib
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from datetime import datetime
|
|
|
|
DB_CONFIG = {
|
|
'host': 'localhost',
|
|
'database': 'adx_system',
|
|
'user': 'admin',
|
|
'password': 'admin123'
|
|
}
|
|
|
|
TRACKING_BASE = "http://151.80.235.110"
|
|
|
|
def get_db():
|
|
return psycopg2.connect(**DB_CONFIG)
|
|
|
|
def get_best_ptr_for_isp(target_isp):
|
|
"""Get best PTR domain for target ISP"""
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT ptr_domain, ip_address, spf_record, reputation_score
|
|
FROM admin.ptr_records
|
|
WHERE %s = ANY(best_target_isps)
|
|
AND status = 'winner'
|
|
AND is_usable = true
|
|
ORDER BY reputation_score DESC, inbox_rate DESC NULLS LAST
|
|
LIMIT 1
|
|
""", (target_isp,))
|
|
|
|
result = cur.fetchone()
|
|
conn.close()
|
|
|
|
if result:
|
|
return {
|
|
'domain': result[0],
|
|
'ip': str(result[1]),
|
|
'spf': result[2],
|
|
'score': result[3]
|
|
}
|
|
return None
|
|
|
|
def get_all_usable_ptrs():
|
|
"""Get all usable PTR domains"""
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT ptr_domain, ip_address, best_target_isps, reputation_score
|
|
FROM admin.ptr_records
|
|
WHERE is_usable = true AND reputation_score >= 60
|
|
ORDER BY reputation_score DESC
|
|
""")
|
|
|
|
results = cur.fetchall()
|
|
conn.close()
|
|
return results
|
|
|
|
def generate_tracking_code(email, ptr_domain):
|
|
"""Generate unique tracking code"""
|
|
data = f"{email}:{ptr_domain}:{datetime.now().timestamp()}"
|
|
return hashlib.md5(data.encode()).hexdigest()[:12]
|
|
|
|
def build_email_ip_direct(to_email, from_domain, subject, body, tracking_code):
|
|
"""Build email for IP Direct sending"""
|
|
|
|
from_email = f"info@{from_domain}"
|
|
|
|
msg = MIMEMultipart('alternative')
|
|
msg['From'] = f"Support <{from_email}>"
|
|
msg['To'] = to_email
|
|
msg['Subject'] = subject
|
|
msg['Message-ID'] = f"<{tracking_code}@{from_domain}>"
|
|
|
|
# Tracking URLs
|
|
click_url = f"{TRACKING_BASE}/c/{tracking_code}"
|
|
pixel_url = f"{TRACKING_BASE}/o/{tracking_code}"
|
|
unsub_url = f"{TRACKING_BASE}/u/{tracking_code}"
|
|
|
|
# Text version
|
|
text_body = body.replace('{click_url}', click_url)
|
|
text_body += f"\n\nUnsubscribe: {unsub_url}"
|
|
|
|
# HTML version
|
|
html_body = f"""<html>
|
|
<body style="font-family: Arial, sans-serif;">
|
|
{body.replace(chr(10), '<br>').replace('{click_url}', f'<a href="{click_url}">Click Here</a>')}
|
|
<br><br>
|
|
<a href="{unsub_url}" style="font-size: 10px; color: #999;">Unsubscribe</a>
|
|
<img src="{pixel_url}" width="1" height="1" style="display:none">
|
|
</body>
|
|
</html>"""
|
|
|
|
msg.attach(MIMEText(text_body, 'plain'))
|
|
msg.attach(MIMEText(html_body, 'html'))
|
|
|
|
return msg, from_email
|
|
|
|
def send_via_pmta(msg, from_email, to_email, pmta_host='127.0.0.1', pmta_port=25):
|
|
"""Send email via PMTA"""
|
|
try:
|
|
with smtplib.SMTP(pmta_host, pmta_port) as server:
|
|
server.sendmail(from_email, to_email, msg.as_string())
|
|
return True
|
|
except Exception as e:
|
|
print(f"Send error: {e}")
|
|
return False
|
|
|
|
def log_send(conn, ptr_domain, to_email, to_isp, tracking_code, success):
|
|
"""Log the send to database"""
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
INSERT INTO admin.seed_tracking
|
|
(seed_email, seed_isp, sender_email, tracking_code, sent_at, delivered)
|
|
VALUES (%s, %s, %s, %s, NOW(), %s)
|
|
""", (to_email, to_isp, f"info@{ptr_domain}", tracking_code, success))
|
|
|
|
# Update PTR stats
|
|
cur.execute("""
|
|
UPDATE admin.ptr_records SET
|
|
total_sent = total_sent + 1,
|
|
last_tested = NOW()
|
|
WHERE ptr_domain = %s
|
|
""", (ptr_domain,))
|
|
|
|
conn.commit()
|
|
|
|
def send_ip_direct(to_email, to_isp, subject, body, ptr_domain=None):
|
|
"""Send email using IP Direct method"""
|
|
|
|
# Get best PTR for this ISP
|
|
if not ptr_domain:
|
|
ptr_info = get_best_ptr_for_isp(to_isp)
|
|
if not ptr_info:
|
|
print(f"No winning PTR found for {to_isp}")
|
|
return False
|
|
ptr_domain = ptr_info['domain']
|
|
print(f"Using PTR: {ptr_domain} (score: {ptr_info['score']})")
|
|
|
|
# Generate tracking
|
|
tracking_code = generate_tracking_code(to_email, ptr_domain)
|
|
|
|
# Build email
|
|
msg, from_email = build_email_ip_direct(to_email, ptr_domain, subject, body, tracking_code)
|
|
|
|
# Send
|
|
success = send_via_pmta(msg, from_email, to_email)
|
|
|
|
# Log
|
|
conn = get_db()
|
|
log_send(conn, ptr_domain, to_email, to_isp, tracking_code, success)
|
|
conn.close()
|
|
|
|
return success
|
|
|
|
def test_ip_direct_method():
|
|
"""Test IP Direct sending to various ISPs"""
|
|
print("=" * 60)
|
|
print("🚀 IP DIRECT SENDER TEST")
|
|
print("=" * 60)
|
|
|
|
# Get available PTRs
|
|
ptrs = get_all_usable_ptrs()
|
|
print(f"\nAvailable PTR domains: {len(ptrs)}")
|
|
|
|
for domain, ip, targets, score in ptrs[:5]:
|
|
targets_str = ', '.join(targets) if targets else 'any'
|
|
print(f" • {domain} ({ip}) - Score: {score} - Targets: {targets_str}")
|
|
|
|
# Test sends (dry run)
|
|
test_targets = [
|
|
('test@ziggo.nl', 'ziggo'),
|
|
('test@kpnmail.nl', 'kpn'),
|
|
('test@hotmail.com', 'hotmail'),
|
|
]
|
|
|
|
print("\n🎯 Best PTR for each ISP:")
|
|
for email, isp in test_targets:
|
|
ptr_info = get_best_ptr_for_isp(isp)
|
|
if ptr_info:
|
|
print(f" {isp}: {ptr_info['domain']} (score: {ptr_info['score']})")
|
|
else:
|
|
print(f" {isp}: No specific PTR, using default")
|
|
|
|
def main():
|
|
import sys
|
|
|
|
if len(sys.argv) > 1:
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd == 'test':
|
|
test_ip_direct_method()
|
|
elif cmd == 'send':
|
|
if len(sys.argv) < 4:
|
|
print("Usage: ip-direct-sender.py send <to_email> <isp> [ptr_domain]")
|
|
return
|
|
to_email = sys.argv[2]
|
|
isp = sys.argv[3]
|
|
ptr = sys.argv[4] if len(sys.argv) > 4 else None
|
|
|
|
subject = "Test IP Direct"
|
|
body = "This is a test email sent via IP Direct method.\n\n{click_url}"
|
|
|
|
success = send_ip_direct(to_email, isp, subject, body, ptr)
|
|
print(f"Send result: {'✅ Success' if success else '❌ Failed'}")
|
|
elif cmd == 'list':
|
|
ptrs = get_all_usable_ptrs()
|
|
print(f"{'Domain':<30} {'IP':<16} {'Score':<6} {'Targets'}")
|
|
print("-" * 70)
|
|
for domain, ip, targets, score in ptrs:
|
|
t = ', '.join(targets[:3]) if targets else '-'
|
|
print(f"{domain:<30} {str(ip):<16} {score:<6} {t}")
|
|
else:
|
|
test_ip_direct_method()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|