File: //usr/local/CyberCP/webmail/webmailManager.py
import json
import os
import base64
from django.http import HttpResponse
from django.shortcuts import render, redirect
from .models import Contact, ContactGroup, ContactGroupMembership, WebmailSettings, SieveRule
from .services.imap_client import IMAPClient
from .services.smtp_client import SMTPClient
from .services.email_composer import EmailComposer
from .services.sieve_client import SieveClient
import plogical.CyberCPLogFileWriter as logging
WEBMAIL_CONF = '/etc/cyberpanel/webmail.conf'
class WebmailManager:
def __init__(self, request):
self.request = request
# ── Helpers ────────────────────────────────────────────────
@staticmethod
def _json_response(data):
return HttpResponse(json.dumps(data), content_type='application/json')
@staticmethod
def _error(msg):
return HttpResponse(json.dumps({'status': 0, 'error_message': str(msg)}),
content_type='application/json')
@staticmethod
def _success(extra=None):
data = {'status': 1}
if extra:
data.update(extra)
return HttpResponse(json.dumps(data), content_type='application/json')
def _get_post_data(self):
try:
return json.loads(self.request.body)
except Exception:
return self.request.POST.dict()
def _get_email(self):
# Check for explicit email in POST body (from account switcher)
# This ensures the correct account is used even if session is stale
try:
data = json.loads(self.request.body)
explicit = data.get('fromAccount', '')
if explicit:
accounts = self._get_managed_accounts()
if explicit in accounts:
self.request.session['webmail_email'] = explicit
return explicit
except Exception:
pass
return self.request.session.get('webmail_email')
def _get_master_config(self):
"""Read master user config from /etc/cyberpanel/webmail.conf"""
try:
with open(WEBMAIL_CONF, 'r') as f:
config = json.load(f)
return config.get('master_user'), config.get('master_password')
except Exception:
return None, None
def _get_imap(self, email_addr=None):
"""Create IMAP client, preferring master user auth for SSO sessions."""
addr = email_addr or self._get_email()
if not addr:
raise Exception('No email account selected')
master_user, master_pass = self._get_master_config()
if master_user and master_pass:
return IMAPClient(addr, '', master_user=master_user, master_password=master_pass)
# Fallback: standalone login with stored password
password = self.request.session.get('webmail_password', '')
return IMAPClient(addr, password)
def _get_smtp(self):
addr = self._get_email()
if not addr:
raise Exception('No email account selected')
# If using master user (SSO), we can't auth to SMTP since
# auth_master_user_separator is not set in Dovecot.
# Use local relay via port 25 instead (Postfix permits localhost).
master_user, master_pass = self._get_master_config()
is_standalone = self.request.session.get('webmail_standalone', False)
if master_user and master_pass and not is_standalone:
return SMTPClient(addr, '', use_local_relay=True)
password = self.request.session.get('webmail_password', '')
return SMTPClient(addr, password)
def _get_sieve(self, email_addr=None):
addr = email_addr or self._get_email()
if not addr:
raise Exception('No email account selected')
master_user, master_pass = self._get_master_config()
if master_user and master_pass:
return SieveClient(addr, '', master_user=master_user, master_password=master_pass)
password = self.request.session.get('webmail_password', '')
return SieveClient(addr, password)
def _get_managed_accounts(self):
"""Get email accounts the current CyberPanel user can access."""
try:
from plogical.acl import ACLManager
from loginSystem.models import Administrator
from mailServer.models import Domains, EUsers
userID = self.request.session['userID']
currentACL = ACLManager.loadedACL(userID)
websites = ACLManager.findAllSites(currentACL, userID)
websites = websites + ACLManager.findChildDomains(websites)
accounts = []
for site in websites:
try:
domain = Domains.objects.get(domain=site)
for eu in EUsers.objects.filter(emailOwner=domain):
accounts.append(eu.email)
except Exception:
continue
return accounts
except Exception:
return []
# ── Page Renders ──────────────────────────────────────────
def loadWebmail(self):
from plogical.httpProc import httpProc
email = self._get_email()
accounts = self._get_managed_accounts()
if not email and accounts:
if len(accounts) == 1:
self.request.session['webmail_email'] = accounts[0]
email = accounts[0]
else:
# Multiple accounts - render picker
proc = httpProc(self.request, 'webmail/index.html',
{'accounts': json.dumps(accounts), 'show_picker': True},
'listEmails')
return proc.render()
proc = httpProc(self.request, 'webmail/index.html',
{'email': email or '',
'accounts': json.dumps(accounts),
'show_picker': False},
'listEmails')
return proc.render()
def loadLogin(self):
return render(self.request, 'webmail/login.html')
# ── Auth APIs ─────────────────────────────────────────────
def apiLogin(self):
data = self._get_post_data()
email_addr = data.get('email', '')
password = data.get('password', '')
if not email_addr or not password:
return self._error('Email and password are required.')
try:
client = IMAPClient(email_addr, password)
client.close()
except Exception as e:
return self._error('Login failed: %s' % str(e))
self.request.session['webmail_email'] = email_addr
self.request.session['webmail_password'] = password
self.request.session['webmail_standalone'] = True
return self._success()
def apiLogout(self):
for key in ['webmail_email', 'webmail_password', 'webmail_standalone']:
self.request.session.pop(key, None)
return self._success()
def apiSSO(self):
"""Auto-login for CyberPanel users."""
accounts = self._get_managed_accounts()
if not accounts:
return self._error('No email accounts found for your user.')
# Preserve previously selected account if still valid
current = self.request.session.get('webmail_email')
if not current or current not in accounts:
current = accounts[0]
self.request.session['webmail_email'] = current
return self._success({'email': current, 'accounts': accounts})
def apiListAccounts(self):
accounts = self._get_managed_accounts()
return self._success({'accounts': accounts})
def apiSwitchAccount(self):
data = self._get_post_data()
email = data.get('email', '')
accounts = self._get_managed_accounts()
if email not in accounts:
return self._error('You do not have access to this account.')
self.request.session['webmail_email'] = email
return self._success({'email': email})
# ── Folder APIs ───────────────────────────────────────────
def apiListFolders(self):
try:
with self._get_imap() as imap:
folders = imap.list_folders()
return self._success({'folders': folders})
except Exception as e:
return self._error(str(e))
def apiCreateFolder(self):
data = self._get_post_data()
name = data.get('name', '')
if not name:
return self._error('Folder name is required.')
try:
with self._get_imap() as imap:
if imap.create_folder(name):
return self._success()
return self._error('Failed to create folder.')
except Exception as e:
return self._error(str(e))
def apiRenameFolder(self):
data = self._get_post_data()
old_name = data.get('oldName', '')
new_name = data.get('newName', '')
if not old_name or not new_name:
return self._error('Both old and new folder names are required.')
try:
with self._get_imap() as imap:
if imap.rename_folder(old_name, new_name):
return self._success()
return self._error('Failed to rename folder.')
except Exception as e:
return self._error(str(e))
def apiDeleteFolder(self):
data = self._get_post_data()
name = data.get('name', '')
if not name:
return self._error('Folder name is required.')
# CyberPanel/Dovecot folder names (INBOX. prefix, separator '.')
protected = ['INBOX', 'INBOX.Sent', 'INBOX.Drafts', 'INBOX.Deleted Items',
'INBOX.Junk E-mail', 'INBOX.Archive']
if name in protected:
return self._error('Cannot delete system folder.')
try:
with self._get_imap() as imap:
if imap.delete_folder(name):
return self._success()
return self._error('Failed to delete folder.')
except Exception as e:
return self._error(str(e))
# ── Message APIs ──────────────────────────────────────────
def apiListMessages(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
page = int(data.get('page', 1))
per_page = int(data.get('perPage', 25))
try:
with self._get_imap() as imap:
result = imap.list_messages(folder, page, per_page)
return self._success(result)
except Exception as e:
return self._error(str(e))
def apiSearchMessages(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
query = data.get('query', '')
try:
with self._get_imap() as imap:
uids = imap.search_messages(folder, query)
uids = [u.decode() if isinstance(u, bytes) else str(u) for u in uids]
return self._success({'uids': uids})
except Exception as e:
return self._error(str(e))
def apiGetMessage(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uid = data.get('uid', '')
if not uid:
return self._error('Message UID is required.')
try:
with self._get_imap() as imap:
msg = imap.get_message(folder, uid)
if msg is None:
return self._error('Message not found.')
imap.mark_read(folder, [uid])
return self._success({'message': msg})
except Exception as e:
return self._error(str(e))
def apiGetAttachment(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uid = data.get('uid', '')
part_id = data.get('partId', '')
try:
with self._get_imap() as imap:
result = imap.get_attachment(folder, uid, part_id)
if result is None:
return self._error('Attachment not found.')
filename, content_type, payload = result
response = HttpResponse(payload, content_type=content_type)
# Sanitize filename to prevent header injection and path traversal
import os as _os
safe_filename = _os.path.basename(filename)
safe_filename = safe_filename.replace('"', '_').replace('\r', '').replace('\n', '').replace('\x00', '')
if not safe_filename:
safe_filename = 'attachment'
response['Content-Disposition'] = 'attachment; filename="%s"' % safe_filename
return response
except Exception as e:
return self._error(str(e))
# ── Action APIs ───────────────────────────────────────────
def apiSendMessage(self):
try:
# For multipart forms, check fromAccount in POST data
if self.request.content_type and 'multipart' in self.request.content_type:
from_account = self.request.POST.get('fromAccount', '')
if from_account:
accounts = self._get_managed_accounts()
if from_account in accounts:
self.request.session['webmail_email'] = from_account
email_addr = self._get_email()
if not email_addr:
return self._error('Not logged in.')
# Handle multipart form data for attachments
if self.request.content_type and 'multipart' in self.request.content_type:
to = self.request.POST.get('to', '')
cc = self.request.POST.get('cc', '')
bcc = self.request.POST.get('bcc', '')
subject = self.request.POST.get('subject', '')
body_html = self.request.POST.get('body', '')
in_reply_to = self.request.POST.get('inReplyTo', '')
references = self.request.POST.get('references', '')
attachments = []
for key in self.request.FILES:
f = self.request.FILES[key]
attachments.append((f.name, f.content_type, f.read()))
else:
data = self._get_post_data()
to = data.get('to', '')
cc = data.get('cc', '')
bcc = data.get('bcc', '')
subject = data.get('subject', '')
body_html = data.get('body', '')
in_reply_to = data.get('inReplyTo', '')
references = data.get('references', '')
attachments = None
if not to:
return self._error('At least one recipient is required.')
mime_msg = EmailComposer.compose(
from_addr=email_addr,
to_addrs=to,
subject=subject,
body_html=body_html,
cc_addrs=cc,
bcc_addrs=bcc,
attachments=attachments,
in_reply_to=in_reply_to,
references=references,
)
smtp = self._get_smtp()
result = smtp.send_message(mime_msg)
if not result['success']:
return self._error(result.get('error', 'Failed to send.'))
# Save to Sent folder
try:
with self._get_imap() as imap:
raw = mime_msg.as_bytes()
smtp.save_to_sent(imap, raw)
except Exception:
pass
# Auto-collect contacts
try:
settings = WebmailSettings.objects.filter(email_account=email_addr).first()
if settings is None or settings.auto_collect_contacts:
self._auto_collect(email_addr, to, cc)
except Exception:
pass
return self._success({'messageId': result['message_id'], 'sentFrom': email_addr})
except Exception as e:
return self._error(str(e))
def _auto_collect(self, owner, to_addrs, cc_addrs=''):
"""Auto-save recipients as contacts."""
import re
all_addrs = '%s,%s' % (to_addrs, cc_addrs) if cc_addrs else to_addrs
emails = re.findall(r'[\w.+-]+@[\w-]+\.[\w.-]+', all_addrs)
for addr in emails:
if addr == owner:
continue
Contact.objects.get_or_create(
owner_email=owner,
email_address=addr,
defaults={'is_auto_collected': True, 'display_name': addr.split('@')[0]},
)
def apiSaveDraft(self):
try:
email_addr = self._get_email()
if not email_addr:
return self._error('Not logged in.')
data = self._get_post_data()
to = data.get('to', '')
subject = data.get('subject', '')
body_html = data.get('body', '')
mime_msg = EmailComposer.compose(
from_addr=email_addr,
to_addrs=to,
subject=subject,
body_html=body_html,
)
with self._get_imap() as imap:
# CyberPanel's Dovecot uses INBOX.Drafts
draft_folders = ['INBOX.Drafts', 'Drafts', 'Draft']
saved = False
for folder in draft_folders:
try:
if imap.append_message(folder, mime_msg.as_bytes(), '\\Draft \\Seen'):
saved = True
break
except Exception:
continue
if not saved:
imap.create_folder('INBOX.Drafts')
imap.append_message('INBOX.Drafts', mime_msg.as_bytes(), '\\Draft \\Seen')
return self._success()
except Exception as e:
return self._error(str(e))
def apiDeleteMessages(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uids = data.get('uids', [])
if not uids:
return self._error('No messages selected.')
try:
with self._get_imap() as imap:
imap.delete_messages(folder, uids)
return self._success()
except Exception as e:
return self._error(str(e))
def apiMoveMessages(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uids = data.get('uids', [])
target = data.get('targetFolder', '')
if not uids or not target:
return self._error('Messages and target folder are required.')
try:
with self._get_imap() as imap:
imap.move_messages(folder, uids, target)
return self._success()
except Exception as e:
return self._error(str(e))
def apiMarkRead(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uids = data.get('uids', [])
try:
with self._get_imap() as imap:
imap.mark_read(folder, uids)
return self._success()
except Exception as e:
return self._error(str(e))
def apiMarkUnread(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uids = data.get('uids', [])
try:
with self._get_imap() as imap:
imap.mark_unread(folder, uids)
return self._success()
except Exception as e:
return self._error(str(e))
def apiMarkFlagged(self):
data = self._get_post_data()
folder = data.get('folder', 'INBOX')
uids = data.get('uids', [])
try:
with self._get_imap() as imap:
imap.mark_flagged(folder, uids)
return self._success()
except Exception as e:
return self._error(str(e))
# ── Contact APIs ──────────────────────────────────────────
def apiListContacts(self):
email = self._get_email()
try:
contacts = list(Contact.objects.filter(owner_email=email).values(
'id', 'display_name', 'email_address', 'phone', 'organization', 'notes', 'is_auto_collected'
))
return self._success({'contacts': contacts})
except Exception as e:
return self._error(str(e))
def apiCreateContact(self):
email = self._get_email()
data = self._get_post_data()
try:
contact = Contact.objects.create(
owner_email=email,
display_name=data.get('displayName', ''),
email_address=data.get('emailAddress', ''),
phone=data.get('phone', ''),
organization=data.get('organization', ''),
notes=data.get('notes', ''),
)
return self._success({'id': contact.id})
except Exception as e:
return self._error(str(e))
def apiUpdateContact(self):
email = self._get_email()
data = self._get_post_data()
contact_id = data.get('id')
try:
contact = Contact.objects.get(id=contact_id, owner_email=email)
for field in ['display_name', 'email_address', 'phone', 'organization', 'notes']:
camel = field.replace('_', ' ').title().replace(' ', '')
camel = camel[0].lower() + camel[1:]
if camel in data:
setattr(contact, field, data[camel])
contact.save()
return self._success()
except Contact.DoesNotExist:
return self._error('Contact not found.')
except Exception as e:
return self._error(str(e))
def apiDeleteContact(self):
email = self._get_email()
data = self._get_post_data()
contact_id = data.get('id')
try:
Contact.objects.filter(id=contact_id, owner_email=email).delete()
return self._success()
except Exception as e:
return self._error(str(e))
def apiSearchContacts(self):
email = self._get_email()
data = self._get_post_data()
query = data.get('query', '')
try:
from django.db.models import Q
contacts = Contact.objects.filter(
owner_email=email
).filter(
Q(display_name__icontains=query) | Q(email_address__icontains=query)
).values('id', 'display_name', 'email_address')[:20]
return self._success({'contacts': list(contacts)})
except Exception as e:
return self._error(str(e))
def apiListContactGroups(self):
email = self._get_email()
try:
groups = list(ContactGroup.objects.filter(owner_email=email).values('id', 'name'))
return self._success({'groups': groups})
except Exception as e:
return self._error(str(e))
def apiCreateContactGroup(self):
email = self._get_email()
data = self._get_post_data()
name = data.get('name', '')
if not name:
return self._error('Group name is required.')
try:
group = ContactGroup.objects.create(owner_email=email, name=name)
return self._success({'id': group.id})
except Exception as e:
return self._error(str(e))
def apiDeleteContactGroup(self):
email = self._get_email()
data = self._get_post_data()
group_id = data.get('id')
try:
ContactGroup.objects.filter(id=group_id, owner_email=email).delete()
return self._success()
except Exception as e:
return self._error(str(e))
# ── Sieve Rule APIs ───────────────────────────────────────
def apiListRules(self):
email = self._get_email()
try:
rules = list(SieveRule.objects.filter(email_account=email).values(
'id', 'name', 'priority', 'is_active',
'condition_field', 'condition_type', 'condition_value',
'action_type', 'action_value',
))
return self._success({'rules': rules})
except Exception as e:
return self._error(str(e))
def apiCreateRule(self):
email = self._get_email()
data = self._get_post_data()
try:
rule = SieveRule.objects.create(
email_account=email,
name=data.get('name', 'New Rule'),
priority=data.get('priority', 0),
is_active=data.get('isActive', True),
condition_field=data.get('conditionField', 'from'),
condition_type=data.get('conditionType', 'contains'),
condition_value=data.get('conditionValue', ''),
action_type=data.get('actionType', 'move'),
action_value=data.get('actionValue', ''),
)
self._sync_sieve_rules(email)
return self._success({'id': rule.id})
except Exception as e:
return self._error(str(e))
def apiUpdateRule(self):
email = self._get_email()
data = self._get_post_data()
rule_id = data.get('id')
try:
rule = SieveRule.objects.get(id=rule_id, email_account=email)
for field in ['name', 'priority', 'is_active', 'condition_field',
'condition_type', 'condition_value', 'action_type', 'action_value']:
camel = field.replace('_', ' ').title().replace(' ', '')
camel = camel[0].lower() + camel[1:]
if camel in data:
val = data[camel]
if field == 'is_active':
val = bool(val)
elif field == 'priority':
val = int(val)
setattr(rule, field, val)
rule.save()
self._sync_sieve_rules(email)
return self._success()
except SieveRule.DoesNotExist:
return self._error('Rule not found.')
except Exception as e:
return self._error(str(e))
def apiDeleteRule(self):
email = self._get_email()
data = self._get_post_data()
rule_id = data.get('id')
try:
SieveRule.objects.filter(id=rule_id, email_account=email).delete()
self._sync_sieve_rules(email)
return self._success()
except Exception as e:
return self._error(str(e))
def apiActivateRules(self):
email = self._get_email()
try:
self._sync_sieve_rules(email)
return self._success()
except Exception as e:
return self._error(str(e))
def _sync_sieve_rules(self, email):
"""Generate sieve script from DB rules and upload to Dovecot.
ManageSieve may not be available if dovecot-sieve/pigeonhole is not
installed or if the ManageSieve service isn't running on port 4190.
Rules are always saved to the database; Sieve sync is best-effort.
"""
rules = SieveRule.objects.filter(email_account=email, is_active=True).order_by('priority')
rule_dicts = []
for r in rules:
rule_dicts.append({
'name': r.name,
'condition_field': r.condition_field,
'condition_type': r.condition_type,
'condition_value': r.condition_value,
'action_type': r.action_type,
'action_value': r.action_value,
})
script = SieveClient.rules_to_sieve(rule_dicts)
try:
with self._get_sieve(email) as sieve:
sieve.put_script('cyberpanel', script)
sieve.activate_script('cyberpanel')
except ConnectionRefusedError:
logging.CyberCPLogFileWriter.writeToFile(
'Sieve sync skipped for %s: ManageSieve not running on port 4190. '
'Install dovecot-sieve and enable ManageSieve.' % email)
except Exception as e:
logging.CyberCPLogFileWriter.writeToFile('Sieve sync failed for %s: %s' % (email, str(e)))
# ── Settings APIs ─────────────────────────────────────────
def apiGetSettings(self):
email = self._get_email()
try:
settings, created = WebmailSettings.objects.get_or_create(email_account=email)
return self._success({
'settings': {
'displayName': settings.display_name,
'signatureHtml': settings.signature_html,
'messagesPerPage': settings.messages_per_page,
'defaultReplyBehavior': settings.default_reply_behavior,
'themePreference': settings.theme_preference,
'autoCollectContacts': settings.auto_collect_contacts,
}
})
except Exception as e:
return self._error(str(e))
def apiSaveSettings(self):
email = self._get_email()
data = self._get_post_data()
try:
settings, created = WebmailSettings.objects.get_or_create(email_account=email)
if 'displayName' in data:
settings.display_name = data['displayName']
if 'signatureHtml' in data:
settings.signature_html = data['signatureHtml']
if 'messagesPerPage' in data:
settings.messages_per_page = int(data['messagesPerPage'])
if 'defaultReplyBehavior' in data:
settings.default_reply_behavior = data['defaultReplyBehavior']
if 'themePreference' in data:
settings.theme_preference = data['themePreference']
if 'autoCollectContacts' in data:
settings.auto_collect_contacts = bool(data['autoCollectContacts'])
settings.save()
return self._success()
except Exception as e:
return self._error(str(e))
# ── Image Proxy ───────────────────────────────────────────
def apiProxyImage(self):
"""Proxy external images to prevent tracking and mixed content."""
if not self._get_email():
return self._error('Not logged in.')
url_b64 = self.request.GET.get('url', '') or self.request.POST.get('url', '')
try:
url = base64.urlsafe_b64decode(url_b64).decode('utf-8')
except Exception:
return self._error('Invalid URL.')
if not url.startswith(('http://', 'https://')):
return self._error('Invalid URL scheme.')
# Block internal/private IPs to prevent SSRF
import urllib.parse
import socket
import ipaddress
hostname = urllib.parse.urlparse(url).hostname or ''
try:
resolved_ip = socket.gethostbyname(hostname)
ip_obj = ipaddress.ip_address(resolved_ip)
if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_multicast or ip_obj.is_reserved:
return self._error('Invalid URL.')
except (socket.gaierror, ValueError):
return self._error('Invalid URL.')
try:
import urllib.request
req = urllib.request.Request(url, headers={
'User-Agent': 'CyberPanel-Webmail-Proxy/1.0',
})
with urllib.request.urlopen(req, timeout=10) as resp:
content_type = resp.headers.get('Content-Type', 'image/png')
if not content_type.startswith('image/'):
return self._error('Not an image.')
data = resp.read(5 * 1024 * 1024) # 5MB max
return HttpResponse(data, content_type=content_type)
except Exception:
return self._error('Failed to fetch image.')