后缀通配符、虚拟别名和收件人分隔符转发故障

后缀通配符、虚拟别名和收件人分隔符转发故障

使用带有 MySQL 后端的 Postfix,设置一组别名,包括域别名。使用“。”作为收件人分隔符,并设置 propagate_unmatched_extensions = canonical, virtual, alias。这就是奇怪的地方。如果我将标记的流量发送到别名[电子邮件保护],我希望 Postfix 将其转发到[电子邮件保护],但是通配符却抓住它并将其标签推送到堆栈上(如果是这种情况的话就应该这样做)。

从发给“用户”的消息中删除 tag2,它会正确地转发给“用户”。发送通配符流量,它会按照预期将标签推送到堆栈上,并将其转发。就好像分隔符短路了别名表扫描,它直接进入通配符搜索。我一直在研究规范文档、论坛等,这个用例有点晦涩难懂,所以如果有人能指出正确的方向来解决这个问题,那就太好了!我猜某个地方的 RE 坏了(不是我的,我所有的垃圾都在 milters 和 SQL 中),但找到它就像大海捞针……

=====更新=====

答案如下,谢谢大家! =====结束更新======

历史原文:

我提炼了一个测试用例如下:

域别名表:
example.net -> machine.example.net

别名表:

@machine.example.net-> 复制代码 [电子邮件保护]

[电子邮件保护]->[电子邮件保护]

一些与别名相关的 SQL...

别名_catchall_maps.cf:

查询 = SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = '%d' 和 alias.address = CONCAT('@',alias_domain.target_domain)AND alias.active = 1 AND alias_domain.active='1'

别名_邮箱_地图.cf:

查询 = 从邮箱、别名域中选择 maildir,其中别名域.别名域 = '%d' 和邮箱.用户名 = CONCAT('%u','@',别名域.target_domain)和邮箱.active = 1 和别名域.active ='1'

别名_域_地图.cf:

查询 = SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = '%d' 和 alias.address = CONCAT('%u', '@', alias_domain.target_domain) AND alias.active = 1 AND alias_domain.active='1'

别名_地图.cf:

查询 = SELECT goto FROM alias WHERE address='%s' AND active = '1'

答案1

更新:添加了测试代码,希望正则表达式转义能够在翻译中存活下来,YMMV......

感谢@ANX 的建议,在 smtpd 和 proxymap 上启用详细日志记录,然后在日志中查找 dict_proxy_lookup 和 input_attribute_value,以了解 postfix 出现问题的位置。测试流量显示,它实际上一直停留在最后一个别名查询(通配符搜索)中,因为它将整个用户名(包括收件人分隔符)传递到查询中。所以我开始查看源代码。果然,在 virtual.c 中,这里解释如下: 大小写折叠 所有投递决定均使用完整的收件人地址(折叠为小写)做出。 表格搜索顺序 通常,查找表被指定为文本文件,用作 postmap 命令的输入。结果为 dbm/db 格式的索引文件,用于邮件系统快速搜索。搜索顺序如下。搜索在第一次成功查找后停止。当收件人具有可选地址扩展名时,[电子邮件保护]首先查找地址。在 <v2.1 中,addr 扩展始终被忽略。 我是一名长期的 Postfix 用户,通过 SQL 代理接口使用虚拟对我来说是一种新体验,因此习惯于从虚拟哈希表中剥离分隔符,这真是一个惊喜。我理解为什么这样做(让人们在规则创建时有选择)。

TL;DR:简而言之,问题如下:在 smptd/virtual.c 中,所有数据库查询都带有接收方分隔符。这使得实施者可以在其 SQL 查询中使用它(或不使用它)。以下是“修复”:在通配符查询之前,将特殊情况查询添加到虚拟别名映射列表中(下面的 mysql_virtual_alias_domain_nodelim_maps.c 摘录),当所有先前的查找都失败但在我们返回到通配符搜索之前,此查询使用正则表达式在查找之前从 %u 中剪切出接收方分隔符:

主文件:

virtual_alias_maps = 代理:mysql:/etc/postfix/sql/mysql_virtual_alias_maps.cf,

代理:mysql:/etc/postfix/sql/mysql_virtual_alias_domain_maps.cf,

代理:mysql:/etc/postfix/sql/mysql_virtual_alias_domain_nodelim_maps.cf,#<--新查询

代理:mysql:/etc/postfix/sql/mysql_virtual_alias_domain_catchall_maps.cf

mysql_virtual_alias_domain_nodelim_maps.cf:(我使用‘.’作为分隔符,YMMV)

查询 = SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = '%d' 和 alias.address = CONCAT(REGEXP_REPLACE('%u', '\..*', ''), '@', alias_domain.target_domain) AND alias.active = 1 AND alias_domain.active='1'

它运行良好,现在可以为 bar.foo 标记的流量设置一个特殊的别名,并且如果别名有自己的标签,它将在标签列表末尾的输出中传播入站标签,这真的很酷!

顺便说一句,我用 Python 构建了一个查询集测试用例,直到满意它的行为与 Postfix 类似。然后开始摆弄正则表达式,直到满意它解决了问题。在解决了 Postfix 和 Python 之间的转义差异后,在非生产 Postfix 实例上尝试了新的查询,它的行为符合预期。

#!/usr/bin/env python3
import mysql.connector
from mysql.connector import Error

# dbhost, dbname, dbuser, dbpass with your parameters
dbhost = ''
dbname = ''
dbuser = ''
dbpass = ''

# Our setup predates DMARC, doing some pretty specialized work with
# address tagging, in particular tying a domain component to the To:
# address as a sub-tag for certain destinations.  This was particularly
# useful in wildcard email hoppers before SPF/DMARC became useful.

# RECIPIENT DELIMITER (. = gmail style)
rd = '.'
# DOMAIN DELIMITER (-)
dd = '-'
# EXAMPLE:
# [email protected]
# where:  tag = folder, or folder.folder, etc.
#         domianlock = site lock parameter

# parseTo(addr, tag_to_push, dom_to_push) 
# breaks an email To: field into constituent parts
# optionally pushes a recipient delimiter and domain
# tag onto the tag stack
def parseTo(to, tag='', dom=''):
    # To: input:
    #                [atag ]
    #         uname.tag-dom@domain
    #[    user             ]
    # uname.tag.tag-dom-dom@domain
    #[             full           ]
    # uname@domain
    #[   notags   ]
    rtv = {}
    to = to.lower()
    if '<' in to:
        to[to.find('<')+1:to.find('>')]
    rtv['user']   = to[:to.find('@')]
    rtv['uname']  = rtv['user']
    rtv['domain'] = to[to.find('@')+1:]
    rtv['atag']   = ''  # inbound tag
    rtv['tag']    = tag  # new tag for top of stack
    rtv['dom']    = dom  # new dom for top of stack
    if rd in rtv['user']:
        rtv['uname']  = rtv['user'][:rtv['user'].find(rd)]
        rtv['atag']  = rtv['user'][rtv['user'].find(rd)+1:]
        if rtv['tag'] != '':
            rtv['tag'] = rd + rtv['tag']
        if dd in rtv['atag']:
            rtv['tag'] = rtv['atag'][:rtv['atag'].find(dd)] + rtv['tag']
            if rtv['dom'] != '':
                rtv['dom'] = dd + rtv['dom']
            rtv['dom'] = rtv['atag'][rtv['atag'].find(dd)+1:] + rtv['dom']
        else:
            rtv['tag']  = rtv['atag'] + rtv['tag']
    rtv['full'] = rtv['uname']
    if rtv['tag'] != '':
        rtv['full'] += rd + rtv['tag']
    if rtv['dom'] != '':
        rtv['full'] += dd + rtv['dom']
    rtv['full'] += '@' + rtv['domain']
    rtv['notags'] = rtv['uname'] + '@' + rtv['domain']
    return rtv

# recurseUser(addr)
# try to match postfix address processing/order as close as possible...
# regex assumes recipient_delimiter = '.'
def recurseUser(addr):
    print("query alias: %s" % (addr))
    to = parseTo(addr)
    user = to['user']
    domain = to['domain']
    # postfix automatically runs the last case (@domain query) to all hashes as
    # well as the list provided to proxymap via the virtual directives in
    # main.cf by re-running the alias case.  Really, it likely will just iterate
    # over all of them but that is a waste given the structure of the virtual db.
    # Cases are run most specific to broadest (e.g. catch-all wildcards).
    # There is a rule specified *somewhere* that alias must contain all valid
    # destinations.  Postfixadmin enforces this for you.
    # try the alias table first with the full spec...
    q = ("SELECT goto FROM alias WHERE address = %s AND active = 1 LIMIT 1")
    cursor.execute(q, (addr,))
    res = cursor.fetchone()
    if res is None:
        # next, check if there is a domain redirect to a specific subdomain alias entry.
        print("  ...not found, check alias domain: %s %s" % (domain, user))
        q = ("SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = %s and alias.address = CONCAT(%s, '@', alias_domain.target_domain) AND alias.active = 1 AND alias_domain.active=1")
        cursor.execute(q, (domain, user))
        res = cursor.fetchone()
        if res is None:
            # ***NEW*** strip any recipient delimiter and try the previous search again
            # this allows the creation of r_d specific aliases, or the use of r_d's with
            # wildcards in the domain cathall maps.  Aliases can also add their own tags,
            # which will be pushed to the top of stack by postix as a matter of course
            print("    ...not found, remove tags (if any) and check again: %s %s" % (domain, to['notags']))
            q = ("SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = %s and alias.address = CONCAT(REGEXP_REPLACE(%s, '\\\\..*', ''), '@', alias_domain.target_domain) AND alias.active = 1 AND alias_domain.active=1")
            cursor.execute(q, (domain, user))
            res = cursor.fetchone()
            if res is None:
                # check for a wildcard query from domain catchall maps (for the cases of [email protected] -> [email protected])
                print("      ...not found, check wc in alias domain: %s" % (domain))
                q = ("SELECT goto FROM alias,alias_domain WHERE alias_domain.alias_domain = %s and alias.address = CONCAT('@', alias_domain.target_domain) AND alias.active = 1 AND alias_domain.active=1")
                cursor.execute(q, (domain,))
                res = cursor.fetchone()
                if res is None:
                    # lastly run the wildcard query for the target wildcard alias (for the cases of [email protected])
                    print("        ... not found, check wc in alias as: @%s" % (domain))
                    q = ("SELECT goto FROM alias WHERE address = %s AND active = 1 LIMIT 1")
                    cursor.execute(q, ('@'+domain,))
                    res = cursor.fetchone()
                    if res is None:
                        print("          Not found, gave up!")
                        return addr
    # found a result, so push any new tags on the stack and recurse until the results are unchanged
    newto = parseTo(res['goto'], to['tag'], to['dom'])
    print("             got: %s" % (res))
    # compare sans tags, we're interested in the actual destination, and the tag stack may change
    # along the way
    if to['notags'] == newto['notags']:
        print("             done!")
        return addr
    print("             recurse...")
    return recurseUser(newto['full']) # otherwise, recurse...


if True:
    db = mysql.connector.connect(database=dbname, host=dbhost, user=dbuser, password=dbpass)
    cursor = db.cursor(dictionary=True)
    while True:
       test = input("Address to try ('q' to end): ")
       if test.lower() == 'q':
          cursor.close()
          db.close()
          quit()
       print("Result:", recurseUser(test))
       print("")

相关内容