/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <Python.h>
#include "qpid/dispatch/python_embedded.h"
#include "policy.h"
#include "policy_internal.h"
#include <stdio.h>
#include <string.h>
#include "dispatch_private.h"
#include "qpid/dispatch/container.h"
#include "qpid/dispatch/server.h"
#include <proton/message.h>
#include <proton/condition.h>
#include <proton/connection.h>
#include <proton/transport.h>
#include <proton/error.h>
#include <proton/event.h>


//
// The current statistics maintained globally through multiple
// reconfiguration of policy settings.
//
static int n_connections = 0;
static int n_denied = 0;
static int n_processed = 0;

//
// error conditions signaled to effect denial
//
static char* RESOURCE_LIMIT_EXCEEDED     = "amqp:resource-limit-exceeded";
//static char* UNAUTHORIZED_ACCESS         = "amqp:unauthorized-access";
//static char* CONNECTION_FORCED           = "amqp:connection:forced";

//
// error descriptions signaled to effect denial
//
static char* CONNECTION_DISALLOWED         = "connection disallowed by local policy";
static char* SESSION_DISALLOWED            = "session disallowed by local policy";
static char* LINK_DISALLOWED               = "link disallowed by local policy";

//
// Policy configuration/statistics management interface
//
struct qd_policy_t {
    qd_dispatch_t        *qd;
    qd_log_source_t      *log_source;
    void                 *py_policy_manager;
                          // configured settings
    int                   max_connection_limit;
    char                 *policyDir;
    bool                  enableVhostPolicy;
                          // live statistics
    int                   connections_processed;
    int                   connections_denied;
    int                   connections_current;
};

/** Create the policy structure
 * @param[in] qd pointer the the qd
 **/
qd_policy_t *qd_policy(qd_dispatch_t *qd)
{
    qd_policy_t *policy = NEW(qd_policy_t);

    policy->qd                   = qd;
    policy->log_source           = qd_log_source("POLICY");
    policy->max_connection_limit = 0;
    policy->policyDir         = 0;
    policy->enableVhostPolicy    = false;
    policy->connections_processed= 0;
    policy->connections_denied   = 0;
    policy->connections_current  = 0;

    qd_log(policy->log_source, QD_LOG_TRACE, "Policy Initialized");
    return policy;
}


/** Free the policy structure
 * @param[in] policy pointer to the policy
 **/
void qd_policy_free(qd_policy_t *policy)
{
    if (policy->policyDir)
        free(policy->policyDir);
    free(policy);
}

//
//
#define CHECK() if (qd_error_code()) goto error

qd_error_t qd_entity_configure_policy(qd_policy_t *policy, qd_entity_t *entity)
{
    policy->max_connection_limit = qd_entity_opt_long(entity, "maxConnections", 0); CHECK();
    if (policy->max_connection_limit < 0)
        return qd_error(QD_ERROR_CONFIG, "maxConnections must be >= 0");
    policy->policyDir =
        qd_entity_opt_string(entity, "policyDir", 0); CHECK();
    policy->enableVhostPolicy = qd_entity_opt_bool(entity, "enableVhostPolicy", false); CHECK();
    qd_log(policy->log_source, QD_LOG_INFO, "Policy configured maxConnections: %d, policyDir: '%s', access rules enabled: '%s'",
           policy->max_connection_limit, policy->policyDir, (policy->enableVhostPolicy ? "true" : "false"));
    return QD_ERROR_NONE;

error:
    if (policy->policyDir)
        free(policy->policyDir);
    qd_policy_free(policy);
    return qd_error_code();
}


//
//
qd_error_t qd_register_policy_manager(qd_policy_t *policy, void *policy_manager)
{
    policy->py_policy_manager = policy_manager;
    return QD_ERROR_NONE;
}


long qd_policy_c_counts_alloc()
{
    qd_policy_denial_counts_t * dc = NEW(qd_policy_denial_counts_t);
    assert(dc);
    memset(dc, 0, sizeof(qd_policy_denial_counts_t));
    return (long)dc;
}


void qd_policy_c_counts_free(long ccounts)
{
    void *dc = (void *)ccounts;
    assert(dc);
    free(dc);
}


qd_error_t qd_policy_c_counts_refresh(long ccounts, qd_entity_t *entity)
{
    qd_policy_denial_counts_t *dc = (qd_policy_denial_counts_t*)ccounts;
    if (!qd_entity_set_long(entity, "sessionDenied", dc->sessionDenied) &&
        !qd_entity_set_long(entity, "senderDenied", dc->senderDenied) &&
        !qd_entity_set_long(entity, "receiverDenied", dc->receiverDenied)
    )
        return QD_ERROR_NONE;
    return qd_error_code();
}


/** Update the statistics in qdrouterd.conf["policy"]
 * @param[in] entity pointer to the policy management object
 **/
qd_error_t qd_entity_refresh_policy(qd_entity_t* entity, void *unused) {
    // Return global stats
    if (!qd_entity_set_long(entity, "connectionsProcessed", n_processed) &&
        !qd_entity_set_long(entity, "connectionsDenied", n_denied) &&
        !qd_entity_set_long(entity, "connectionsCurrent", n_connections)
    )
        return QD_ERROR_NONE;
    return qd_error_code();
}


//
// Functions related to absolute connection counts.
// These handle connections at the socket level with
// no regard to user identity. Simple yes/no decisions
// are made and there is no AMQP channel for returning
// error conditions.
//

bool qd_policy_socket_accept(void *context, const char *hostname)
{
    qd_policy_t *policy = (qd_policy_t *)context;
    bool result = true;

    if (policy->max_connection_limit == 0) {
        // Policy not in force; connection counted and allowed
        n_connections += 1;
    } else {
        // Policy in force
        if (n_connections < policy->max_connection_limit) {
            // connection counted and allowed
            n_connections += 1;
            qd_log(policy->log_source, QD_LOG_TRACE, "ALLOW Connection '%s' based on global connection count. N= %d", hostname, n_connections);
        } else {
            // connection denied
            result = false;
            n_denied += 1;
            qd_log(policy->log_source, QD_LOG_INFO, "DENY Connection '%s' based on global connection count. N= %d", hostname, n_connections);
        }
    }
    n_processed += 1;
    return result;
}


//
//
void qd_policy_socket_close(void *context, const qd_connection_t *conn)
{
    qd_policy_t *policy = (qd_policy_t *)context;

    n_connections -= 1;
    assert (n_connections >= 0);
    if (policy->enableVhostPolicy) {
        // HACK ALERT: TODO: This should be deferred to a Python thread
        qd_python_lock_state_t lock_state = qd_python_lock();
        PyObject *module = PyImport_ImportModule("qpid_dispatch_internal.policy.policy_manager");
        if (module) {
            PyObject *close_connection = PyObject_GetAttrString(module, "policy_close_connection");
            if (close_connection) {
                PyObject *result = PyObject_CallFunction(close_connection, "(OK)",
                                                         (PyObject *)policy->py_policy_manager,
                                                          conn->connection_id);
                if (result) {
                    Py_XDECREF(result);
                } else {
                    qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: Connection close failed: result");
                }
                Py_XDECREF(close_connection);
            } else {
                qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: Connection close failed: close_connection");
            }
            Py_XDECREF(module);
        } else {
            qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: Connection close failed: module");
        }
        qd_python_unlock(lock_state);
    }
    if (policy->max_connection_limit > 0) {
        const char *hostname = qdpn_connector_name(conn->pn_cxtr);
        qd_log(policy->log_source, QD_LOG_DEBUG, "Connection '%s' closed with resources n_sessions=%d, n_senders=%d, n_receivers=%d. N= %d.",
                hostname, conn->n_sessions, conn->n_senders, conn->n_receivers, n_connections);
    }
}


//
// Functions related to authenticated connection denial.
// An AMQP Open has been received over some connection.
// Evaluate the connection auth and the Open fields to
// allow or deny the Open. Denied Open attempts are
// effected by returning Open and then Close_with_condition.
//
/** Look up user/host/app in python vhost and give the AMQP Open
 *  a go-no_go decision. Return false if the mechanics of calling python
 *  fails. A policy lookup will deny the connection by returning a blank
 *  usergroup name in the name buffer.
 *  Connection and connection denial counting is done in the python code.
 * @param[in] policy pointer to policy
 * @param[in] username authenticated user name
 * @param[in] hostip numeric host ip address
 * @param[in] app application name received in remote AMQP Open.hostname
 * @param[in] conn_name connection name for tracking
 * @param[out] name_buf pointer to settings name buffer
 * @param[in] name_buf_size size of settings_buf
 **/
bool qd_policy_open_lookup_user(
    qd_policy_t *policy,
    const char *username,
    const char *hostip,
    const char *app,
    const char *conn_name,
    char       *name_buf,
    int         name_buf_size,
    uint64_t    conn_id,
    qd_policy_settings_t *settings)
{
    // TODO: crolke 2016-03-24 - Workaround for PROTON-1133: Port number is included in Open hostname
    // Strip the ':NNNN', if any, from the app name so that policy will work with proton 0.12
    char appname[HOST_NAME_MAX + 1];
    strncpy(appname, app, HOST_NAME_MAX);
    appname[HOST_NAME_MAX] = 0;
    char * colonp = strstr(appname, ":");
    if (colonp) {
        *colonp = 0;
    }
    // Lookup the user/host/app for allow/deny and to get settings name
    bool res = false;
    qd_python_lock_state_t lock_state = qd_python_lock();
    PyObject *module = PyImport_ImportModule("qpid_dispatch_internal.policy.policy_manager");
    if (module) {
        PyObject *lookup_user = PyObject_GetAttrString(module, "policy_lookup_user");
        if (lookup_user) {
            PyObject *result = PyObject_CallFunction(lookup_user, "(OssssK)",
                                                     (PyObject *)policy->py_policy_manager,
                                                     username, hostip, appname, conn_name, conn_id);
            if (result) {
                const char *res_string = PyString_AsString(result);
                strncpy(name_buf, res_string, name_buf_size);
                Py_XDECREF(result);
                res = true; // settings name returned
            } else {
                qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_user: result");
            }
            Py_XDECREF(lookup_user);
        } else {
            qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_user: lookup_user");
        }
    }
    if (!res) {
        if (module) {
            Py_XDECREF(module);
        }
        qd_python_unlock(lock_state);
        return false;
    }

    // 
    if (name_buf[0]) {
        // Go get the named settings
        res = false;
        PyObject *upolicy = PyDict_New();
        if (upolicy) {
            PyObject *lookup_settings = PyObject_GetAttrString(module, "policy_lookup_settings");
            if (lookup_settings) {
                PyObject *result2 = PyObject_CallFunction(lookup_settings, "(OssO)",
                                                        (PyObject *)policy->py_policy_manager,
                                                        appname, name_buf, upolicy);
                if (result2) {
                    settings->maxFrameSize         = qd_entity_opt_long((qd_entity_t*)upolicy, "maxFrameSize", 0);
                    settings->maxMessageSize       = qd_entity_opt_long((qd_entity_t*)upolicy, "maxMessageSize", 0);
                    settings->maxSessionWindow     = qd_entity_opt_long((qd_entity_t*)upolicy, "maxSessionWindow", 0);
                    settings->maxSessions          = qd_entity_opt_long((qd_entity_t*)upolicy, "maxSessions", 0);
                    settings->maxSenders           = qd_entity_opt_long((qd_entity_t*)upolicy, "maxSenders", 0);
                    settings->maxReceivers         = qd_entity_opt_long((qd_entity_t*)upolicy, "maxReceivers", 0);
                    settings->allowAnonymousSender = qd_entity_opt_bool((qd_entity_t*)upolicy, "allowAnonymousSender", false);
                    settings->allowDynamicSrc      = qd_entity_opt_bool((qd_entity_t*)upolicy, "allowDynamicSrc", false);
                    settings->sources              = qd_entity_get_string((qd_entity_t*)upolicy, "sources");
                    settings->targets              = qd_entity_get_string((qd_entity_t*)upolicy, "targets");
                    settings->denialCounts         = (qd_policy_denial_counts_t*)
                                                    qd_entity_get_long((qd_entity_t*)upolicy, "denialCounts");
                    Py_XDECREF(result2);
                    res = true; // named settings content returned
                } else {
                    qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_user: result2");
                }
                Py_XDECREF(lookup_settings);
            } else {
                qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_user: lookup_settings");
            }
            Py_XDECREF(upolicy);
        } else {
            qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_user: upolicy");
        }
    }
    Py_XDECREF(module);
    qd_python_unlock(lock_state);

    if (name_buf[0]) {
        qd_log(policy->log_source,
           QD_LOG_TRACE,
           "ALLOW AMQP Open lookup_user: %s, hostip: %s, app: %s, connection: %s. Usergroup: '%s'%s",
           username, hostip, appname, conn_name, name_buf, (res ? "" : " Internal error."));
    } else {
        // Denials are logged in python code
    }

    return res;
}


//
//
void qd_policy_private_deny_amqp_connection(pn_connection_t *conn, const char *cond_name, const char *cond_descr)
{
    pn_condition_t * cond = pn_connection_condition(conn);
    (void) pn_condition_set_name(       cond, cond_name);
    (void) pn_condition_set_description(cond, cond_descr);
    pn_connection_close(conn);
    // Connection denial counts are counted and logged by python code.
}


//
//
void qd_policy_deny_amqp_session(pn_session_t *ssn, qd_connection_t *qd_conn)
{
    pn_condition_t * cond = pn_session_condition(ssn);
    (void) pn_condition_set_name(       cond, RESOURCE_LIMIT_EXCEEDED);
    (void) pn_condition_set_description(cond, SESSION_DISALLOWED);
    pn_session_close(ssn);
    qd_conn->policy_settings->denialCounts->sessionDenied++;
}


//
//
bool qd_policy_approve_amqp_session(pn_session_t *ssn, qd_connection_t *qd_conn)
{
    bool result = true;
    if (qd_conn->policy_settings) {
        if (qd_conn->policy_settings->maxSessions) {
            if (qd_conn->n_sessions == qd_conn->policy_settings->maxSessions) {
                qd_policy_deny_amqp_session(ssn, qd_conn);
                result = false;
            }
        }
    }
    pn_connection_t *conn = qd_connection_pn(qd_conn);
    qd_dispatch_t *qd = qd_conn->server->qd;
    qd_policy_t *policy = qd->policy;
    const char *hostip = qdpn_connector_hostip(qd_conn->pn_cxtr);
    const char *app = pn_connection_remote_hostname(conn);
    if (result) {
        qd_log(policy->log_source,
           QD_LOG_TRACE,
           "ALLOW AMQP Begin Session. user: %s, hostip: %s, app: %s",
           qd_conn->user_id, hostip, app);
    } else {
        qd_log(policy->log_source,
           QD_LOG_INFO,
           "DENY AMQP Begin Session due to session limit. user: %s, hostip: %s, app: %s",
           qd_conn->user_id, hostip, app);
    }
    return result;
}


//
//
void qd_policy_apply_session_settings(pn_session_t *ssn, qd_connection_t *qd_conn)
{
    if (qd_conn->policy_settings && qd_conn->policy_settings->maxSessionWindow) {
        pn_session_set_incoming_capacity(ssn, qd_conn->policy_settings->maxSessionWindow);
    } else {
        pn_session_set_incoming_capacity(ssn, 1000000);
    }
}

//
//
void _qd_policy_deny_amqp_link(pn_link_t *link, qd_connection_t *qd_conn)
{
    pn_condition_t * cond = pn_link_condition(link);
    (void) pn_condition_set_name(       cond, RESOURCE_LIMIT_EXCEEDED);
    (void) pn_condition_set_description(cond, LINK_DISALLOWED);
    pn_link_close(link);
}


//
//
void _qd_policy_deny_amqp_sender_link(pn_link_t *pn_link, qd_connection_t *qd_conn)
{
    _qd_policy_deny_amqp_link(pn_link, qd_conn);
    qd_conn->policy_settings->denialCounts->senderDenied++;
}


//
//
void _qd_policy_deny_amqp_receiver_link(pn_link_t *pn_link, qd_connection_t *qd_conn)
{
    _qd_policy_deny_amqp_link(pn_link, qd_conn);
    qd_conn->policy_settings->denialCounts->receiverDenied++;
}


//
//
#define MIN(a,b) (((a)<(b))?(a):(b))

char * _qd_policy_link_user_name_subst(const char *uname, const char *proposed, char *obuf, int osize)
{
    if (strlen(uname) == 0)
        return NULL;

    const char *duser = "${user}";
    char *retptr = obuf;
    const char *wiptr = proposed;
    const char *findptr = strstr(proposed, uname);
    if (findptr == NULL) {
        return NULL;
    }

    // Copy leading before match
    int segsize = findptr - wiptr;
    int copysize = MIN(osize, segsize);
    if (copysize)
        strncpy(obuf, wiptr, copysize);
    wiptr += copysize;
    osize -= copysize;
    obuf  += copysize;

    // Copy the substitution string
    segsize = strlen(duser);
    copysize = MIN(osize, segsize);
    if (copysize)
        strncpy(obuf, duser, copysize);
    wiptr += strlen(uname);
    osize -= copysize;
    obuf  += copysize;

    // Copy trailing after match
    strncpy(obuf, wiptr, osize);
    return retptr;
}


//
//
// Size of 'easy' temporary copy of allowed input string
#define QPALN_SIZE 1024
// Size of user-name-substituted proposed string.
#define QPALN_USERBUFSIZE 300
// C in the CSV string
#define QPALN_COMMA_SEP ","
// Wildcard character
#define QPALN_WILDCARD '*'

bool _qd_policy_approve_link_name(const char *username, const char *allowed, const char *proposed)
{
    // Verify string sizes are usable
    size_t p_len = strlen(proposed);
    if (p_len == 0) {
        // degenerate case of blank name being opened. will never match anything.
        return false;
    }
    size_t a_len = strlen(allowed);
    if (a_len == 0) {
        // no names in 'allowed'.
        return false;
    }

    // Create a temporary writable copy of incoming allowed list
    char t_allow[QPALN_SIZE + 1]; // temporary buffer for normal allow lists
    char * pa = t_allow;
    if (a_len > QPALN_SIZE) {
        pa = (char *)malloc(a_len + 1); // malloc a buffer for larger allow lists
    }
    strncpy(pa, allowed, a_len);
    pa[a_len] = 0;
    // Do reverse user substitution into proposed
    char substbuf[QPALN_USERBUFSIZE];
    char * prop2 = _qd_policy_link_user_name_subst(username, proposed, substbuf, QPALN_USERBUFSIZE);
    char *toknext = 0;
    char *tok = strtok_r(pa, QPALN_COMMA_SEP, &toknext);
    assert (tok);
    bool result = false;
    while (tok != NULL) {
        if (*tok == QPALN_WILDCARD) {
            result = true;
            break;
        }
        int matchlen = p_len;
        int len = strlen(tok);
        if (tok[len-1] == QPALN_WILDCARD) {
            matchlen = len - 1;
            assert(len > 0);
        }
        if (strncmp(tok, proposed, matchlen) == 0) {
            result = true;
            break;
        }
        if (prop2 && strncmp(tok, prop2, matchlen) == 0) {
            result = true;
            break;
        }
        tok = strtok_r(NULL, QPALN_COMMA_SEP, &toknext);
    }
    if (pa != t_allow) {
        free(pa);
    }
    return result;
}

//
//
bool qd_policy_approve_amqp_sender_link(pn_link_t *pn_link, qd_connection_t *qd_conn)
{
    const char *hostip = qdpn_connector_hostip(qd_conn->pn_cxtr);
    const char *app = pn_connection_remote_hostname(qd_connection_pn(qd_conn));

    if (qd_conn->policy_settings->maxSenders) {
        if (qd_conn->n_senders == qd_conn->policy_settings->maxSenders) {
            // Max sender limit specified and violated.
            qd_log(qd_conn->server->qd->policy->log_source, QD_LOG_INFO,
                "DENY AMQP Attach sender for user '%s', host '%s', app '%s' based on maxSenders limit",
                qd_conn->user_id, hostip, app);
            _qd_policy_deny_amqp_sender_link(pn_link, qd_conn);
            return false;
        } else {
            // max sender limit not violated
        }
    } else {
        // max sender limit not specified
    }
    // Approve sender link based on target
    const char * target = pn_terminus_get_address(pn_link_remote_target(pn_link));
    bool lookup;
    if (target && *target) {
        // a target is specified
        lookup = _qd_policy_approve_link_name(qd_conn->user_id, qd_conn->policy_settings->targets, target);

        qd_log(qd_conn->server->qd->policy->log_source, (lookup ? QD_LOG_TRACE : QD_LOG_INFO),
            "%s AMQP Attach sender link '%s' for user '%s', host '%s', app '%s' based on link target name",
            (lookup ? "ALLOW" : "DENY"), target, qd_conn->user_id, hostip, app);

        if (!lookup) {
            _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
            return false;
        }
    } else {
        // A sender with no remote target.
        // This happens all the time with anonymous relay
        lookup = qd_conn->policy_settings->allowAnonymousSender;
        qd_log(qd_conn->server->qd->policy->log_source, (lookup ? QD_LOG_TRACE : QD_LOG_INFO),
            "%s AMQP Attach anonymous sender for user '%s', host '%s', app '%s'",
            (lookup ? "ALLOW" : "DENY"), qd_conn->user_id, hostip, app);
        if (!lookup) {
            _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
            return false;
        }
    }
    // Approved
    return true;
}


bool qd_policy_approve_amqp_receiver_link(pn_link_t *pn_link, qd_connection_t *qd_conn)
{
    const char *hostip = qdpn_connector_hostip(qd_conn->pn_cxtr);
    const char *app = pn_connection_remote_hostname(qd_connection_pn(qd_conn));

    if (qd_conn->policy_settings->maxReceivers) {
        if (qd_conn->n_receivers == qd_conn->policy_settings->maxReceivers) {
            // Max sender limit specified and violated.
            qd_log(qd_conn->server->qd->policy->log_source, QD_LOG_INFO,
                "DENY AMQP Attach receiver for user '%s', host '%s', app '%s' based on maxReceivers limit",
                qd_conn->user_id, hostip, app);
            _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
            return false;
        } else {
            // max receiver limit not violated
        }
    } else {
        // max receiver limit not specified
    }
    // Approve receiver link based on source
    bool dynamic_src = pn_terminus_is_dynamic(pn_link_remote_source(pn_link));
    if (dynamic_src) {
        bool lookup = qd_conn->policy_settings->allowDynamicSrc;
        qd_log(qd_conn->server->qd->policy->log_source, (lookup ? QD_LOG_TRACE : QD_LOG_INFO),
            "%s AMQP Attach receiver dynamic source for user '%s', host '%s', app '%s',",
            (lookup ? "ALLOW" : "DENY"), qd_conn->user_id, hostip, app);
        // Dynamic source policy rendered the decision
        if (!lookup) {
            _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
        }
        return lookup;
    }
    const char * source = pn_terminus_get_address(pn_link_remote_source(pn_link));
    if (source && *source) {
        // a source is specified
        bool lookup = _qd_policy_approve_link_name(qd_conn->user_id, qd_conn->policy_settings->sources, source);

        qd_log(qd_conn->server->qd->policy->log_source, (lookup ? QD_LOG_TRACE : QD_LOG_INFO),
            "%s AMQP Attach receiver link '%s' for user '%s', host '%s', app '%s' based on link source name",
            (lookup ? "ALLOW" : "DENY"), source, qd_conn->user_id, hostip, app);

        if (!lookup) {
            _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
            return false;
        }
    } else {
        // A receiver with no remote source.
        qd_log(qd_conn->server->qd->policy->log_source, QD_LOG_TRACE,
               "DENY AMQP Attach receiver link '' for user '%s', host '%s', app '%s'",
               qd_conn->user_id, hostip, app);
        _qd_policy_deny_amqp_receiver_link(pn_link, qd_conn);
        return false;
    }
    // Approved
    return true;
}


//
//
void qd_policy_amqp_open(void *context, bool discard)
{
    qd_connection_t *qd_conn = (qd_connection_t *)context;
    if (!discard) {
        pn_connection_t *conn = qd_connection_pn(qd_conn);
        qd_dispatch_t *qd = qd_conn->server->qd;
        qd_policy_t *policy = qd->policy;
        bool connection_allowed = true;

        if (policy->enableVhostPolicy) {
            // Open connection or not based on policy.
            pn_transport_t *pn_trans = pn_connection_transport(conn);
            const char *hostip = qdpn_connector_hostip(qd_conn->pn_cxtr);
            const char *pcrh = pn_connection_remote_hostname(conn);
            const char *app = (pcrh ? pcrh : "");
            const char *conn_name = qdpn_connector_name(qd_conn->pn_cxtr);
#define SETTINGS_NAME_SIZE 256
            char settings_name[SETTINGS_NAME_SIZE];
            uint32_t conn_id = qd_conn->connection_id;
            qd_conn->policy_settings = NEW(qd_policy_settings_t); // TODO: memory pool for settings
            memset(qd_conn->policy_settings, 0, sizeof(qd_policy_settings_t));

            if (qd_policy_open_lookup_user(policy, qd_conn->user_id, hostip, app, conn_name,
                                           settings_name, SETTINGS_NAME_SIZE, conn_id,
                                           qd_conn->policy_settings) &&
                settings_name[0]) {
                // This connection is allowed by policy.
                // Apply transport policy settings
                if (qd_conn->policy_settings->maxFrameSize > 0)
                    pn_transport_set_max_frame(pn_trans, qd_conn->policy_settings->maxFrameSize);
                if (qd_conn->policy_settings->maxSessions > 0)
                    pn_transport_set_channel_max(pn_trans, qd_conn->policy_settings->maxSessions - 1);
            } else {
                // This connection is denied by policy.
                connection_allowed = false;
            }
        } else {
            // No policy implies automatic policy allow
            // Note that connections not governed by policy have no policy_settings.
        }
        if (connection_allowed) {
            if (pn_connection_state(conn) & PN_LOCAL_UNINIT)
                pn_connection_open(conn);
            policy_notify_opened(qd_conn->open_container, qd_conn, qd_conn->context);
        } else {
            qd_policy_private_deny_amqp_connection(conn, RESOURCE_LIMIT_EXCEEDED, CONNECTION_DISALLOWED);
        }
    }
    qd_connection_set_event_stall(qd_conn, false);
}