RC9/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java

/*
 *  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.
 *
 *
 */

package org.apache.qpid.server.security.access.plugins;

import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicConsumeBody;
import org.apache.qpid.framing.BasicPublishBody;

import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.security.access.ACLManager;
import org.apache.qpid.server.security.access.ACLPlugin;
import org.apache.qpid.server.security.access.AccessResult;
import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.security.access.PrincipalPermissions;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * This uses the default
 */
public class SimpleXML implements ACLPlugin
{
    private Map<String, PrincipalPermissions> _users;
    private final AccessResult GRANTED = new AccessResult(this, AccessResult.AccessStatus.GRANTED);

    public SimpleXML()
    {
        _users = new ConcurrentHashMap<String, PrincipalPermissions>();
    }

    public void setConfiguaration(Configuration config)
    {
        processConfig(config);
    }

    private void processConfig(Configuration config)
    {
        processPublish(config);

        processConsume(config);

        processCreate(config);
    }

    /**
     * Publish format takes
     * Exchange + Routing Key Pairs
     *
     * @param config XML Configuration
     */
    private void processPublish(Configuration config)
    {
        Configuration publishConfig = config.subset("security.access_control_list.publish");

        //Process users that have full publish permission
        String[] users = publishConfig.getStringArray("users.user");

        for (String user : users)
        {
            grant(Permission.PUBLISH, user);
        }

        // Process exchange limited users
        int exchangeCount = 0;
        Configuration exchangeConfig = publishConfig.subset("exchanges.exchange(" + exchangeCount + ")");

        while (!exchangeConfig.isEmpty())
        {
            //Get Exchange Name
            AMQShortString exchangeName = new AMQShortString(exchangeConfig.getString("name"));

            //Get Routing Keys
            int keyCount = 0;
            Configuration routingkeyConfig = exchangeConfig.subset("routing_keys.routing_key(" + keyCount + ")");

            while (!routingkeyConfig.isEmpty())
            {
                //Get RoutingKey Value
                AMQShortString routingKeyValue = new AMQShortString(routingkeyConfig.getString("value"));

                //Apply Exchange + RoutingKey permissions to Users
                users = routingkeyConfig.getStringArray("users.user");
                for (String user : users)
                {
                    grant(Permission.PUBLISH, user, exchangeName, routingKeyValue);
                }

                //Apply permissions to Groups

                // Check for more configs
                keyCount++;
                routingkeyConfig = exchangeConfig.subset("routing_keys.routing_key(" + keyCount + ")");
            }

            //Apply Exchange wide permissions to Users
            users = exchangeConfig.getStringArray("exchange(" + exchangeCount + ").users.user");

            for (String user : users)
            {
                grant(Permission.PUBLISH, user, exchangeName);
            }

            //Apply permissions to Groups
            exchangeCount++;
            exchangeConfig = publishConfig.subset("exchanges.exchange(" + exchangeCount + ")");
        }
    }

    private void grant(Permission permission, String user, Object... parameters)
    {
        PrincipalPermissions permissions = _users.get(user);

        if (permissions == null)
        {
            permissions = new PrincipalPermissions(user);
        }

        _users.put(user, permissions);
        permissions.grant(permission, parameters);
    }

    private void processConsume(Configuration config)
    {
        Configuration consumeConfig = config.subset("security.access_control_list.consume");

        // Process queue limited users
        int queueCount = 0;
        Configuration queueConfig = consumeConfig.subset("queues.queue(" + queueCount + ")");

        while (!queueConfig.isEmpty())
        {
            //Get queue Name
            AMQShortString queueName = new AMQShortString(queueConfig.getString("name"));
            // if there is no name then there may be a temporary element
            boolean temporary = queueConfig.containsKey("temporary");
            boolean ownQueues = queueConfig.containsKey("own_queues");

            //Process permissions for this queue
            String[] users = queueConfig.getStringArray("users.user");
            for (String user : users)
            {
                grant(Permission.CONSUME, user, queueName, temporary, ownQueues);
            }

            //See if we have another config
            queueCount++;
            queueConfig = consumeConfig.subset("queues.queue(" + queueCount + ")");
        }

        // Process users that have full consume permission
        String[] users = consumeConfig.getStringArray("users.user");

        for (String user : users)
        {
            grant(Permission.CONSUME, user);
        }
    }

    private void processCreate(Configuration config)
    {
        Configuration createConfig = config.subset("security.access_control_list.create");

        // Process create permissions for queue creation
        int queueCount = 0;
        Configuration queueConfig = createConfig.subset("queues.queue(" + queueCount + ")");

        while (!queueConfig.isEmpty())
        {
            //Get queue Name
            AMQShortString queueName = new AMQShortString(queueConfig.getString("name"));

            // if there is no name then there may be a temporary element
            boolean temporary = queueConfig.containsKey("temporary");

            int exchangeCount = 0;
            Configuration exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");

            while (!exchangeConfig.isEmpty())
            {

                AMQShortString exchange = new AMQShortString(exchangeConfig.getString("name"));
                AMQShortString routingKey = new AMQShortString(exchangeConfig.getString("routing_key"));

                //Process permissions for this queue
                String[] users = exchangeConfig.getStringArray("users.user");
                for (String user : users)
                {
                    grant(Permission.CREATE, user, temporary,
                          (queueName.equals("") ? null : queueName),
                          (exchange.equals("") ? null : exchange),
                          (routingKey.equals("") ? null : routingKey));
                }

                //See if we have another config
                exchangeCount++;
                exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");
            }

            // Process users that are not bound to an exchange
            String[] users = queueConfig.getStringArray("users.user");

            for (String user : users)
            {
                grant(Permission.CREATE, user, temporary, queueName);
            }

            //See if we have another config
            queueCount++;
            queueConfig = createConfig.subset("queues.queue(" + queueCount + ")");
        }

        // Process create permissions for exchange creation
        int exchangeCount = 0;
        Configuration exchangeConfig = createConfig.subset("exchanges.exchange(" + exchangeCount + ")");

        while (!exchangeConfig.isEmpty())
        {
            AMQShortString exchange = new AMQShortString(exchangeConfig.getString("name"));
            AMQShortString clazz = new AMQShortString(exchangeConfig.getString("class"));

            //Process permissions for this queue
            String[] users = exchangeConfig.getStringArray("users.user");
            for (String user : users)
            {
                grant(Permission.CREATE, user, exchange, clazz);
            }

            //See if we have another config
            exchangeCount++;
            exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");
        }

        // Process users that have full create permission
        String[] users = createConfig.getStringArray("users.user");

        for (String user : users)
        {
            grant(Permission.CREATE, user);
        }


    }

    public String getPluginName()
    {
        return "Simple";
    }

    public AccessResult authorise(AMQProtocolSession session, Permission permission, AMQMethodBody body, Object... parameters) throws AMQConnectionException
    {
        String error = "";
        
        if (ACLManager.getLogger().isInfoEnabled())
        {
            ACLManager.getLogger().info("Simple Authorisation processing user:" + session.getAuthorizedID() + " for :" + permission.toString()
                    + " on " + body.getClass().getSimpleName()
                    + (parameters == null || parameters.length == 0 ? "" : "-" + AllowAll.accessablesToString(parameters)));
        }

        String username = session.getAuthorizedID().getName();

        //Get the Users Permissions
        PrincipalPermissions permissions = _users.get(username);

        if (permissions != null)
        {
            switch (permission)
            {
                case ACCESS:
                    return GRANTED;
                case BIND:  // Body QueueDeclareBody - Parameters : Exchange, Queue, QueueName
                    // Body QueueBindBody - Paramters : Exchange, Queue, QueueName
                    if (parameters.length == 3)
                    {
                        // Parameters : Exchange, Queue, RoutingKey
                        if (permissions.authorise(Permission.BIND, body, parameters[0], parameters[1], parameters[2]))
                        {
                            return GRANTED;
                        }
                    }
                    break;
                case CONSUME: // Parameters : none
                    if (parameters.length == 1 && permissions.authorise(Permission.CONSUME, parameters[0]))
                    {
                        return GRANTED;
                    }
                    break;
                case CREATE: // Body : QueueDeclareBody | ExchangeDeclareBody - Parameters : none
                    if (permissions.authorise(Permission.CREATE, body))
                    {
                        return GRANTED;
                    }
                    break;
                case PUBLISH: // Body : BasicPublishBody  Parameters : exchange
                    if (parameters.length == 1 && parameters[0] instanceof Exchange)
                    {
                        if (permissions.authorise(Permission.PUBLISH, ((Exchange) parameters[0]).getName(),
                                                  ((BasicPublishBody) body).getRoutingKey()))
                        {
                            return GRANTED;
                        }
                    }
                    break;
                case PURGE:
                    break;
                case DELETE:
                    break;
                case UNBIND:
                    break;
            }
        }

        //todo potential refactor this ConnectionException Out of here
        throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, error);
    }
}