diff options
| author | Aidan Skinner <aidan@apache.org> | 2008-02-28 17:31:56 +0000 |
|---|---|---|
| committer | Aidan Skinner <aidan@apache.org> | 2008-02-28 17:31:56 +0000 |
| commit | d99d23e76493ea70d33020fd761b33d59a724f29 (patch) | |
| tree | f0cfdf8b806d4b219a4c7facb497fb1cd2f8f368 | |
| parent | 603bff3fda0e820072b5309af0d15f2a7f049737 (diff) | |
| download | qpid-python-d99d23e76493ea70d33020fd761b33d59a724f29.tar.gz | |
Merged revisions 630239,630247,630723-630725,630728-630731,630733,630833,630854,630857,630865,630872-630873,630892,630896,631141,631264,631295-631296,631651,631936 via svnmerge from
https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1
........
r630239 | rgodfrey | 2008-02-22 16:15:11 +0000 (Fri, 22 Feb 2008) | 1 line
QPID-790 : Performance Improvements
........
r630247 | aidan | 2008-02-22 16:50:26 +0000 (Fri, 22 Feb 2008) | 2 lines
QPID-771: fix up exception handling a bit more, this is so lovely.
........
r630723 | ritchiem | 2008-02-25 03:22:39 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-806 : Added startsWith and endsWith methods to AMQShortString, including test.
........
r630724 | ritchiem | 2008-02-25 03:33:58 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-807 : made methods public.
........
r630725 | ritchiem | 2008-02-25 03:36:55 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-808 : add method to qpid.jms Interface.
........
r630728 | ritchiem | 2008-02-25 03:52:54 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-809 : Added comments and additional error logging.
........
r630729 | ritchiem | 2008-02-25 03:58:25 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-107 added ACCESS_REFUSED as a reason to throw AMQAuthenticationException
........
r630730 | ritchiem | 2008-02-25 03:59:58 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-809 : Added comments and improved logging msgs.
........
r630731 | ritchiem | 2008-02-25 04:06:32 +0000 (Mon, 25 Feb 2008) | 1 line
Adjusted the default debug level to debug (it was only info)
........
r630733 | ritchiem | 2008-02-25 04:56:42 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-810 : Moved check for closingChannels higher in stack and close channel on any AMQException being thrown from the body.handle methods.
........
r630833 | ritchiem | 2008-02-25 12:59:03 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-811 : The RejectionEE is occuring because the task pool is shutdown before the close-ok has been received.
........
r630854 | ritchiem | 2008-02-25 13:54:46 +0000 (Mon, 25 Feb 2008) | 6 lines
QPID-107 : Initial ACL implementation for review. Implemented Permissions : Consume, Create, Publish.
The Permissions are configured via XML in a user friendly way.
Sections for consume, create and publish are currently used to further define Access and Bind internally.
Access is granted to all users that have some permission. Bind rights are given to users with Create rights.
Full details of the ACL design will be posted on the wiki :
http://cwiki.apache.org/qpid/qpid-design-access-control-lists.html
........
r630857 | ritchiem | 2008-02-25 14:11:06 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-107 : Missing acl example config
........
r630865 | ritchiem | 2008-02-25 14:50:04 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-809 Updated logging and removed an unnecessary printStackTrace().
........
r630872 | ritchiem | 2008-02-25 15:16:01 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-809 Updated logging and some whitespace changes.
........
r630873 | ritchiem | 2008-02-25 15:18:23 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-809 Updated logging and some whitespace changes.
........
r630892 | ritchiem | 2008-02-25 15:52:24 +0000 (Mon, 25 Feb 2008) | 1 line
QPID-107 : Fixed the left over ACL'd ApplicationRegistry from killing subsequent tests.
........
r630896 | ritchiem | 2008-02-25 15:58:38 +0000 (Mon, 25 Feb 2008) | 2 lines
QPID-809 Updated ignore list
........
r631141 | ritchiem | 2008-02-26 09:15:31 +0000 (Tue, 26 Feb 2008) | 1 line
QPID=813 : Synchronized getInstance
........
r631264 | rgodfrey | 2008-02-26 15:55:29 +0000 (Tue, 26 Feb 2008) | 1 line
QPID-804 : Fix Java Broker Python test failures
........
r631295 | rupertlssmith | 2008-02-26 16:57:05 +0000 (Tue, 26 Feb 2008) | 1 line
QPID-800 : Fixed copyright notices that I had previously missed.
........
r631296 | ritchiem | 2008-02-26 16:57:29 +0000 (Tue, 26 Feb 2008) | 1 line
QPID-107 : Fixed the left over old AccessManager/PrincipalDatabase references.
........
r631651 | ritchiem | 2008-02-27 17:11:45 +0000 (Wed, 27 Feb 2008) | 1 line
Updated ignore list for eclipse files.
........
r631936 | rgodfrey | 2008-02-28 12:05:16 +0000 (Thu, 28 Feb 2008) | 1 line
QPID-818 : Persistent Pub/Sub can get exception on acking message
........
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/thegreatmerge@632068 13f79535-47bb-0310-9956-ffa450edef68
81 files changed, 3651 insertions, 1778 deletions
diff --git a/qpid/java/broker/etc/acl.config.xml b/qpid/java/broker/etc/acl.config.xml new file mode 100644 index 0000000000..4de8168fcc --- /dev/null +++ b/qpid/java/broker/etc/acl.config.xml @@ -0,0 +1,229 @@ +<?xml version="1.0" encoding="ISO-8859-1"?> +<!-- + - + - 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. + - + --> +<broker> + <prefix>${QPID_HOME}</prefix> + <work>${QPID_WORK}</work> + <conf>${prefix}/etc</conf> + <connector> + <!-- Uncomment out this block and edit the keystorePath and keystorePassword + to enable SSL support + <ssl> + <enabled>true</enabled> + <sslOnly>true</sslOnly> + <keystorePath>/path/to/keystore.ks</keystorePath> + <keystorePassword>keystorepass</keystorePassword> + </ssl>--> + <qpidnio>false</qpidnio> + <!-- I've had the 0.0 and 0.1 Reader threads continually throwing IOException when client closes--> + <protectio>false</protectio> + <transport>nio</transport> + <port>5672</port> + <sslport>8672</sslport> + <socketReceiveBuffer>32768</socketReceiveBuffer> + <socketSendBuffer>32768</socketSendBuffer> + </connector> + <management> + <enabled>true</enabled> + <jmxport>8999</jmxport> + <security-enabled>false</security-enabled> + </management> + <advanced> + <filterchain enableExecutorPool="true"/> + <enablePooledAllocator>false</enablePooledAllocator> + <enableDirectBuffers>false</enableDirectBuffers> + <framesize>65535</framesize> + <compressBufferOnQueue>false</compressBufferOnQueue> + <enableJMSXUserID>true</enableJMSXUserID> + </advanced> + + <security> + <principal-databases> + <!-- Example use of Base64 encoded MD5 hashes for authentication via CRAM-MD5-Hashed --> + <principal-database> + <name>passwordfile</name> + <class>org.apache.qpid.server.security.auth.database.Base64MD5PasswordFilePrincipalDatabase</class> + <attributes> + <attribute> + <name>passwordFile</name> + <value>${conf}/qpid.passwd</value> + </attribute> + </attributes> + </principal-database> + </principal-databases> + + <access> + <class>org.apache.qpid.server.security.access.plugins.DenyAll</class> + </access> + + <jmx> + <access>${conf}/jmxremote.access</access> + <principal-database>passwordfile</principal-database> + </jmx> + </security> + + <virtualhosts> + <directory>${conf}/virtualhosts</directory> + + <virtualhost> + <name>test</name> + <test> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + + <queues> + <exchange>amq.direct</exchange> + <!-- 4Mb --> + <maximumQueueDepth>4235g264</maximumQueueDepth> + <!-- 2Mb --> + <maximumMessageSize>2117632</maximumMessageSize> + <!-- 10 mins --> + <maximumMessageAge>600000</maximumMessageAge> + </queues> + + + <security> + <access> + <class>org.apache.qpid.server.security.access.plugins.SimpleXML</class> + </access> + + <access_control_list> + <!-- This section grants pubish rights to an exchange + routing key pair --> + <publish> + <exchanges> + <exchange> + <name>amq.direct</name> + <routing_keys> + + <!-- Allow clients to publish requests --> + <routing_key> + <value>example.RequestQueue</value> + <users> + <user>client</user> + </users> + </routing_key> + + <!-- Allow the processor to respond to a client on their Temporary Topic --> + <routing_key> + <value>tmp_*</value> + <users> + <user>server</user> + </users> + </routing_key> + </routing_keys> + + </exchange> + </exchanges> + </publish> + + <!-- This section grants users the ability to consume from the broker --> + <consume> + <queues> + + <!-- Allow the clients to consume from their temporary queues--> + <queue> + <temporary/> + <users> + <user>client</user> + </users> + </queue> + + + <!-- Only allow the server to consume from the Request Queue--> + <queue> + <name>example.RequestQueue</name> + <users> + <user>server</user> + </users> + </queue> + + + </queues> + </consume> + + <!-- This section grants clients the ability to create queues and exchanges --> + <create> + <queues> + <!-- Allow clients to create temporary queues--> + <queue> + <temporary/> + <exchanges> + <exchange> + <name>amq.direct</name> + <users> + <user>client</user> + </users> + </exchange> + </exchanges> + </queue> + <!-- Allow the server to create the Request Queue--> + <queue> + <name>example.RequestQueue</name> + <users> + <user>server</user> + </users> + </queue> + + </queues> + </create> + + + </access_control_list> + + </security> + </test> + </virtualhost> + + + <virtualhost> + <name>development</name> + <development> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + </development> + </virtualhost> + + <virtualhost> + <name>localhost</name> + <localhost> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + </localhost> + </virtualhost> + + </virtualhosts> + + <heartbeat> + <delay>0</delay> + <timeoutFactor>2.0</timeoutFactor> + </heartbeat> + + <queue> + <auto_register>false</auto_register> + </queue> + + <virtualhosts>${conf}/virtualhosts.xml</virtualhosts> +</broker> + + diff --git a/qpid/java/broker/etc/config.xml b/qpid/java/broker/etc/config.xml index 0009f7d255..6c650b95eb 100644 --- a/qpid/java/broker/etc/config.xml +++ b/qpid/java/broker/etc/config.xml @@ -55,32 +55,21 @@ <security> <principal-databases> + <!-- Example use of Base64 encoded MD5 hashes for authentication via CRAM-MD5-Hashed --> <principal-database> <name>passwordfile</name> - <class>org.apache.qpid.server.security.auth.database.PlainPasswordVhostFilePrincipalDatabase</class> + <class>org.apache.qpid.server.security.auth.database.PlainPasswordFilePrincipalDatabase</class> <attributes> <attribute> <name>passwordFile</name> - <value>${conf}/passwdVhost</value> + <value>${conf}/passwd</value> </attribute> </attributes> </principal-database> - - <!-- Example use of Base64 encoded MD5 hashes for authentication via CRAM-MD5-Hashed - <principal-database> - <name>passwordfile</name> - <class>org.apache.qpid.server.security.auth.database.Base64MD5PasswordFilePrincipalDatabase</class> - <attributes> - <attribute> - <name>passwordFile</name> - <value>${conf}/qpid.passwd</value> - </attribute> - </attributes> - </principal-database--> </principal-databases> <access> - <class>org.apache.qpid.server.security.access.AllowAll</class> + <class>org.apache.qpid.server.security.access.plugins.AllowAll</class> </access> <jmx> <access>${conf}/jmxremote.access</access> @@ -89,13 +78,12 @@ </security> <virtualhosts> + <directory>${conf}/virtualhosts</directory> + <virtualhost> <name>localhost</name> <localhost> <store> - <!-- <class>org.apache.qpid.server.store.berkeleydb.BDBMessageStore</class> - <environment-path>${work}/localhost-store</environment-path> --> - <class>org.apache.qpid.server.store.MemoryMessageStore</class> </store> <txn> @@ -108,27 +96,6 @@ <expiredMessageCheckPeriod>20000</expiredMessageCheckPeriod> </housekeeping> - <security> - <!-- Need protocol changes to allow this--> - <authentication> - <name>passwordfile</name> - <!-- Currently this can't be used as Vhost isn't specified at connection start only connection open --> - <mechanism>PLAIN</mechanism> - </authentication> - <access> - <class>org.apache.qpid.server.security.access.PrincipalDatabaseAccessManager</class> - <attributes> - <attribute> - <name>principalDatabase</name> - <value>passwordfile</value> - </attribute> - <attribute> - <name>defaultAccessManager</name> - <value>DenyAll</value> - </attribute> - </attributes> - </access> - </security> </localhost> </virtualhost> @@ -141,11 +108,6 @@ <txn> <class>org.apache.qpid.server.txn.MemoryTransactionManager</class> </txn> - <security> - <name>passwordfile-notusedyet</name> - <mechanism>PLAIN</mechanism> - <mechanism>CRAM-MD5</mechanism> - </security> </development> </virtualhost> @@ -158,21 +120,6 @@ <txn> <class>org.apache.qpid.server.txn.MemoryTransactionManager</class> </txn> - <security> - <name>passwordfile-notusedyet</name> - <mechanism>PLAIN</mechanism> - <mechanism>CRAM-MD5</mechanism> - </security> - <access> - <class>org.apache.qpid.server.security.access.PrincipalDatabaseAccessManager</class> - <attributes> - <attribute> - <name>principalDatabase</name> - <value>rubbish-to-cause-default</value> - </attribute> - </attributes> - </access> - </test> </virtualhost> @@ -189,4 +136,3 @@ </broker> - diff --git a/qpid/java/broker/etc/debug.log4j.xml b/qpid/java/broker/etc/debug.log4j.xml index e8fd7e119d..71f9502b75 100644 --- a/qpid/java/broker/etc/debug.log4j.xml +++ b/qpid/java/broker/etc/debug.log4j.xml @@ -106,9 +106,9 @@ <!-- Log all info events to file --> <root> - <priority value="info"/> + <priority value="debug"/> <appender-ref ref="STDOUT"/> - <appender-ref ref="FileAppender"/> + <!--appender-ref ref="FileAppender"/--> </root> </log4j:configuration> diff --git a/qpid/java/broker/etc/passwd b/qpid/java/broker/etc/passwd index 966a16153d..7aca438551 100644 --- a/qpid/java/broker/etc/passwd +++ b/qpid/java/broker/etc/passwd @@ -17,3 +17,6 @@ # under the License. # guest:guest +client:guest +server:guest + diff --git a/qpid/java/broker/etc/persistent_config.xml b/qpid/java/broker/etc/persistent_config.xml index ed0591256b..5b3151e6fc 100644 --- a/qpid/java/broker/etc/persistent_config.xml +++ b/qpid/java/broker/etc/persistent_config.xml @@ -51,18 +51,18 @@ <principal-databases> <principal-database> <name>passwordfile</name> - <class>org.apache.qpid.server.security.auth.database.PlainPasswordVhostFilePrincipalDatabase</class> + <class>org.apache.qpid.server.security.auth.database.PlainPasswordFilePrincipalDatabase</class> <attributes> <attribute> <name>passwordFile</name> - <value>${conf}/passwdVhost</value> + <value>${conf}/passwd</value> </attribute> </attributes> </principal-database> </principal-databases> <access> - <class>org.apache.qpid.server.security.access.AllowAll</class> + <class>org.apache.qpid.server.security.access.plugins.AllowAll</class> </access> <jmx> <access>${conf}/jmxremote.access</access> @@ -76,7 +76,7 @@ <localhost> <store> <environment-path>${work}/bdbstore/localhost-store</environment-path> - <class>org.apache.qpid.server.store.berkeleydb.messageStore.MessageStoreImpl</class> + <class>org.apache.qpid.server.store.berkeleydb.messageStore.MessageStoreImpl</class> </store> <txn> <environment-tx-timeout>60</environment-tx-timeout> diff --git a/qpid/java/broker/etc/qpid.passwd b/qpid/java/broker/etc/qpid.passwd index 79b5e11777..dbfb9d1923 100644 --- a/qpid/java/broker/etc/qpid.passwd +++ b/qpid/java/broker/etc/qpid.passwd @@ -19,4 +19,5 @@ guest:CE4DQ6BIb/BVMN9scFyLtA==
admin:ISMvKXpXpadDiUoOSoAfww==
user:CE4DQ6BIb/BVMN9scFyLtA==
-
+server:CE4DQ6BIb/BVMN9scFyLtA==
+client:CE4DQ6BIb/BVMN9scFyLtA==
diff --git a/qpid/java/broker/etc/transient_config.xml b/qpid/java/broker/etc/transient_config.xml index ddd5203ccb..fcffd90ba2 100644 --- a/qpid/java/broker/etc/transient_config.xml +++ b/qpid/java/broker/etc/transient_config.xml @@ -61,7 +61,7 @@ </principal-database> </principal-databases> <access> - <class>org.apache.qpid.server.security.access.AllowAll</class> + <class>org.apache.qpid.server.security.access.plugins.AllowAll</class> </access> <jmx> <access>${conf}/jmxremote.access</access> @@ -84,7 +84,7 @@ <security> <access> - <class>org.apache.qpid.server.security.access.PrincipalDatabaseAccessManager</class> + <class>org.apache.qpid.server.security.old.PrincipalDatabaseAccessManager</class> <attributes> <attribute> <name>principalDatabase</name> diff --git a/qpid/java/broker/src/main/java/log4j.properties b/qpid/java/broker/src/main/java/log4j.properties index 87f04f4991..6788c65463 100644 --- a/qpid/java/broker/src/main/java/log4j.properties +++ b/qpid/java/broker/src/main/java/log4j.properties @@ -19,6 +19,6 @@ log4j.rootCategory=${amqj.logging.level}, console log4j.appender.console=org.apache.log4j.ConsoleAppender -log4j.appender.console.Threshold=info +log4j.appender.console.Threshold=all log4j.appender.console.layout=org.apache.log4j.PatternLayout log4j.appender.console.layout.ConversionPattern=%t %d %p [%c{4}] %m%n diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java index ac29998c2a..c62a7880a8 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java @@ -100,10 +100,9 @@ public class TxAck implements TxnOp //make persistent changes, i.e. dequeue and decrementReference for (UnacknowledgedMessage msg : _unacked) { - //msg.restoreTransientMessageData(); - //Message has been ack so discard it. This will dequeue and decrement the reference. msg.discard(storeContext); + } } @@ -115,7 +114,6 @@ public class TxAck implements TxnOp //in memory (persistent changes will be rolled back by store) for (UnacknowledgedMessage msg : _unacked) { - msg.clearTransientMessageData(); msg.getMessage().takeReference(); } } @@ -124,11 +122,6 @@ public class TxAck implements TxnOp { //remove the unacked messages from the channels map _map.remove(_unacked); - for (UnacknowledgedMessage msg : _unacked) - { - msg.clearTransientMessageData(); - } - } public void rollback(StoreContext storeContext) diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java index 40f5970cac..df7cecc940 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java @@ -68,16 +68,6 @@ public class UnacknowledgedMessage entry.getMessage().decrementReference(storeContext); } - public void restoreTransientMessageData() throws AMQException - { - entry.getMessage().restoreTransientMessageData(); - } - - public void clearTransientMessageData() - { - entry.getMessage().clearTransientMessageData(); - } - public AMQMessage getMessage() { return entry.getMessage(); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java index adeb61ccbb..cfb90ae449 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java @@ -86,7 +86,7 @@ public class DestWildExchange extends AbstractExchange new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); private final ConcurrentHashMap<AMQShortString, List<AMQQueue>> _wildCardBindingKey2queues = new ConcurrentHashMap<AMQShortString, List<AMQQueue>>(); - // private ConcurrentHashMap<AMQShortString, AMQQueue> _routingKey2queue = new ConcurrentHashMap<AMQShortString, AMQQueue>(); + private static final byte TOPIC_SEPARATOR = (byte)'.'; private static final AMQShortString TOPIC_SEPARATOR_AS_SHORTSTRING = new AMQShortString("."); private static final AMQShortString AMQP_STAR_TOKEN = new AMQShortString("*"); @@ -279,16 +279,6 @@ public class DestWildExchange extends AbstractExchange AMQShortString normalizedString = AMQShortString.join(subscriptionList, TOPIC_SEPARATOR_AS_SHORTSTRING); -/* - StringBuilder sb = new StringBuilder(); - for (AMQShortString s : subscriptionList) - { - sb.append(s); - sb.append(TOPIC_SEPARATOR); - } - - sb.deleteCharAt(sb.length() - 1); -*/ return normalizedString; } @@ -459,11 +449,6 @@ public class DestWildExchange extends AbstractExchange { AMQShortString next = routingTokens.nextToken(); - /* if (next.equals(AMQP_HASH) && routingkeyTokens.get(routingkeyTokens.size() - 1).equals(AMQP_HASH)) - { - continue; - } - */ routingkeyTokens[token++] = next; } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java index 2061803d65..32f58ed666 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilter.java @@ -56,7 +56,7 @@ public class JMSSelectorFilter implements MessageFilter catch (AMQException e) { //fixme this needs to be sorted.. it shouldn't happen - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } return false; } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java index b3fc752ba6..4e816b03c2 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java @@ -7,9 +7,9 @@ * 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 @@ -24,20 +24,20 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.ConsumerTagNotUniqueException; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.ExistingExclusiveSubscriptionException; import org.apache.qpid.server.queue.ExistingSubscriptionPreventsExclusiveException; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; public class BasicConsumeMethodHandler implements StateAwareMethodListener<BasicConsumeBody> { - private static final Logger _log = Logger.getLogger(BasicConsumeMethodHandler.class); + private static final Logger _logger = Logger.getLogger(BasicConsumeMethodHandler.class); private static final BasicConsumeMethodHandler _instance = new BasicConsumeMethodHandler(); @@ -67,21 +67,21 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic } else { - if (_log.isDebugEnabled()) + if (_logger.isDebugEnabled()) { - _log.debug("BasicConsume: from '" + body.getQueue() + - "' for:" + body.getConsumerTag() + - " nowait:" + body.getNowait() + - " args:" + body.getArguments()); + _logger.debug("BasicConsume: from '" + body.getQueue() + + "' for:" + body.getConsumerTag() + + " nowait:" + body.getNowait() + + " args:" + body.getArguments()); } AMQQueue queue = body.getQueue() == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.getQueue().intern()); if (queue == null) { - if (_log.isDebugEnabled()) + if (_logger.isDebugEnabled()) { - _log.debug("No queue for '" + body.getQueue() + "'"); + _logger.debug("No queue for '" + body.getQueue() + "'"); } if (body.getQueue() != null) { @@ -99,6 +99,9 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic final AMQShortString consumerTagName; + //Perform ACLs + vHost.getAccessManager().authorise(session, Permission.CONSUME, body, queue); + if (body.getConsumerTag() != null) { consumerTagName = body.getConsumerTag().intern(); @@ -125,7 +128,7 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic } catch (org.apache.qpid.AMQInvalidArgumentException ise) { - _log.debug("Closing connection due to invalid selector"); + _logger.debug("Closing connection due to invalid selector"); MethodRegistry methodRegistry = session.getMethodRegistry(); AMQMethodBody responseBody = methodRegistry.createChannelCloseBody(AMQConstant.INVALID_ARGUMENT.getCode(), diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java index 8d69697350..f8f9127809 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java @@ -27,10 +27,10 @@ import org.apache.qpid.framing.BasicGetBody; import org.apache.qpid.framing.BasicGetEmptyBody;
import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.security.access.Permission;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -82,7 +82,11 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB }
else
{
- if(!queue.performGet(session, channel, !body.getNoAck()))
+
+ //Perform ACLs
+ vHost.getAccessManager().authorise(session, Permission.CONSUME, body, queue);
+
+ if (!queue.performGet(session, channel, !body.getNoAck()))
{
MethodRegistry methodRegistry = session.getMethodRegistry();
// TODO - set clusterId
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java index eafea9b6a6..16b53820fd 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java @@ -25,20 +25,19 @@ import org.apache.qpid.AMQException; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.BasicPublishBody; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.MethodRegistry; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.framing.abstraction.MessagePublishInfo; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; -public class BasicPublishMethodHandler implements StateAwareMethodListener<BasicPublishBody> +public class BasicPublishMethodHandler implements StateAwareMethodListener<BasicPublishBody> { - private static final Logger _log = Logger.getLogger(BasicPublishMethodHandler.class); + private static final Logger _logger = Logger.getLogger(BasicPublishMethodHandler.class); private static final BasicPublishMethodHandler _instance = new BasicPublishMethodHandler(); @@ -55,12 +54,9 @@ public class BasicPublishMethodHandler implements StateAwareMethodListener<Basi public void methodReceived(AMQStateManager stateManager, BasicPublishBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - - - - if (_log.isDebugEnabled()) + if (_logger.isDebugEnabled()) { - _log.debug("Publish received on channel " + channelId); + _logger.debug("Publish received on channel " + channelId); } AMQShortString exchange = body.getExchange(); @@ -90,6 +86,9 @@ public class BasicPublishMethodHandler implements StateAwareMethodListener<Basi throw body.getChannelNotFoundException(channelId); } + //Access Control + vHost.getAccessManager().authorise(session, Permission.PUBLISH, body, e); + MessagePublishInfo info = session.getMethodRegistry().getProtocolVersionMethodConverter().convertToInfo(body); info.setExchange(exchange); channel.setPublishFrame(info, session, e); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java index 4f62e0b930..f99e650979 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java @@ -23,14 +23,12 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQState; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.security.access.AccessResult; -import org.apache.qpid.server.security.access.AccessRights; import org.apache.log4j.Logger; public class ConnectionOpenMethodHandler implements StateAwareMethodListener<ConnectionOpenBody> @@ -79,22 +77,8 @@ public class ConnectionOpenMethodHandler implements StateAwareMethodListener<Con { session.setVirtualHost(virtualHost); - AccessResult result = virtualHost.getAccessManager().isAuthorized(virtualHost, session.getAuthorizedID(), AccessRights.Rights.ANY); - - switch (result.getStatus()) - { - default: - case REFUSED: - String error = "Any access denied to vHost '" + virtualHostName + "' by " - + result.getAuthorizer(); - - _logger.warn(error); - - throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, error); - case GRANTED: - _logger.info("Granted any access to vHost '" + virtualHostName + "' for " + session.getAuthorizedID() - + " by '" + result.getAuthorizer() + "'"); - } + //Perform ACL + virtualHost.getAccessManager().authorise(session, Permission.ACCESS ,body, virtualHost); // See Spec (0.8.2). Section 3.1.2 Virtual Hosts if (session.getContextKey() == null) diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java index be6030e6fc..dee6a98802 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java @@ -26,11 +26,11 @@ import org.apache.qpid.AMQException; import org.apache.qpid.AMQUnknownExchangeType; import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeFactory; import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -58,7 +58,9 @@ public class ExchangeDeclareHandler implements StateAwareMethodListener<Exchange VirtualHost virtualHost = session.getVirtualHost(); ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory(); - + + //Perform ACL + virtualHost.getAccessManager().authorise(session, Permission.CREATE, body); if (_logger.isDebugEnabled()) { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java index 8f36e6c767..888ffcb2e5 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java @@ -21,13 +21,12 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.ExchangeDeleteBody; import org.apache.qpid.framing.ExchangeDeleteOkBody; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.exchange.ExchangeInUseException; import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -51,6 +50,9 @@ public class ExchangeDeleteHandler implements StateAwareMethodListener<ExchangeD VirtualHost virtualHost = session.getVirtualHost(); ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); + //Perform ACLs + virtualHost.getAccessManager().authorise(session, Permission.DELETE,body, + exchangeRegistry.getExchange(body.getExchange())); try { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java index a365cd864a..0f6dc7a19d 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java @@ -25,13 +25,13 @@ import org.apache.qpid.AMQException; import org.apache.qpid.AMQInvalidRoutingKeyException; import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -106,6 +106,10 @@ public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody> try { + + //Perform ACLs + virtualHost.getAccessManager().authorise(session, Permission.BIND, body, exch, queue, routingKey); + if (!exch.isBound(routingKey, body.getArguments(), queue)) { queue.bind(routingKey, body.getArguments(), exch); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java index 5dabd75ed2..8ef14e64a8 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java @@ -7,9 +7,9 @@ * 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 @@ -26,10 +26,9 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.configuration.Configured; -import org.apache.qpid.exchange.ExchangeDefaults; + import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.configuration.Configurator; import org.apache.qpid.server.configuration.VirtualHostConfiguration; import org.apache.qpid.server.exchange.Exchange; @@ -37,6 +36,7 @@ import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.security.access.Permission; import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener; import org.apache.qpid.server.store.MessageStore; @@ -46,7 +46,7 @@ import org.apache.commons.configuration.Configuration; public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclareBody> { - private static final Logger _log = Logger.getLogger(QueueDeclareHandler.class); + private static final Logger _logger = Logger.getLogger(QueueDeclareHandler.class); private static final QueueDeclareHandler _instance = new QueueDeclareHandler(); @@ -74,6 +74,8 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); MessageStore store = virtualHost.getMessageStore(); + // Perform ACL on queue Creation + virtualHost.getAccessManager().authorise(session, Permission.CREATE, body); @@ -106,7 +108,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar } else { - queue = createQueue(queueName,body, virtualHost, session); + queue = createQueue(queueName, body, virtualHost, session); if (queue.isDurable() && !queue.isAutoDelete()) { //DTX MessageStore @@ -123,8 +125,12 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar { Exchange defaultExchange = exchangeRegistry.getDefaultExchange(); + // Perform ACL to control bindings + virtualHost.getAccessManager().authorise(session, Permission.BIND, body, + defaultExchange, queue, queueName); + queue.bind(queueName, null, defaultExchange); - _log.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getName() + ")"); + _logger.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getName() + ")"); } } } @@ -156,7 +162,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar queue.getConsumerCount()); session.writeFrame(responseBody.generateFrame(channelId)); - _log.info("Queue " + queueName + " declared successfully"); + _logger.info("Queue " + queueName + " declared successfully"); } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java index 76d3d5d936..02efedc27e 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java @@ -36,6 +36,7 @@ import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.exception.InternalErrorException; import org.apache.qpid.server.exception.QueueDoesntExistException; +import org.apache.qpid.server.security.access.Permission; public class QueueDeleteHandler implements StateAwareMethodListener<QueueDeleteBody> { @@ -105,6 +106,10 @@ public class QueueDeleteHandler implements StateAwareMethodListener<QueueDeleteB } else { + + //Perform ACLs + virtualHost.getAccessManager().authorise(session, Permission.DELETE, body, queue); + int purged = queue.delete(body.getIfUnused(), body.getIfEmpty()); if (queue.isDurable()) diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java index 5bdca93bc6..cce49f13c7 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java @@ -35,6 +35,7 @@ import org.apache.qpid.server.state.AMQStateManager; import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.security.access.Permission;
public class QueuePurgeHandler implements StateAwareMethodListener<QueuePurgeBody>
{
@@ -100,6 +101,10 @@ public class QueuePurgeHandler implements StateAwareMethodListener<QueuePurgeBod }
else
{
+
+ //Perform ACLs
+ virtualHost.getAccessManager().authorise(session, Permission.PURGE, body, queue);
+
long purged = queue.clearQueue(channel.getStoreContext());
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java index b056fa6797..e758e315aa 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java @@ -1,110 +1,113 @@ -package org.apache.qpid.server.handler;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.framing.*;
-import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
-import org.apache.qpid.server.state.StateAwareMethodListener;
-import org.apache.qpid.server.state.AMQStateManager;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.virtualhost.VirtualHost;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQInvalidRoutingKeyException;
-import org.apache.qpid.protocol.AMQConstant;
-
-public class QueueUnbindHandler implements StateAwareMethodListener<QueueUnbindBody>
-{
- private static final Logger _log = Logger.getLogger(QueueUnbindHandler.class);
-
- private static final QueueUnbindHandler _instance = new QueueUnbindHandler();
-
- public static QueueUnbindHandler getInstance()
- {
- return _instance;
- }
-
- private QueueUnbindHandler()
- {
- }
-
- public void methodReceived(AMQStateManager stateManager, QueueUnbindBody body, int channelId) throws AMQException
- {
- AMQProtocolSession session = stateManager.getProtocolSession();
- VirtualHost virtualHost = session.getVirtualHost();
- ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();
- QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
-
-
- final AMQQueue queue;
- final AMQShortString routingKey;
-
- if (body.getQueue() == null)
- {
- AMQChannel channel = session.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId);
- }
-
- queue = channel.getDefaultQueue();
-
- if (queue == null)
- {
- throw body.getConnectionException(AMQConstant.NOT_FOUND, "No default queue defined on channel and queue was null");
- }
-
- routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern();
-
- }
- else
- {
- queue = queueRegistry.getQueue(body.getQueue());
- routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern();
- }
-
- if (queue == null)
- {
- throw body.getConnectionException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist.");
- }
- final Exchange exch = exchangeRegistry.getExchange(body.getExchange());
- if (exch == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.getExchange() + " does not exist.");
- }
-
-
- try
- {
- queue.unBind(routingKey, body.getArguments(), exch);
- }
- catch (AMQInvalidRoutingKeyException rke)
- {
- throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString());
- }
- catch (AMQException e)
- {
- if(e.getErrorCode() == AMQConstant.NOT_FOUND)
- {
- throw body.getConnectionException(AMQConstant.NOT_FOUND,e.getMessage(),e);
- }
- throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString());
- }
-
- if (_log.isInfoEnabled())
- {
- _log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey);
- }
-
- MethodRegistry_0_9 methodRegistry = (MethodRegistry_0_9) session.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createQueueUnbindOkBody();
- session.writeFrame(responseBody.generateFrame(channelId));
-
-
- }
-}
+package org.apache.qpid.server.handler; + +import org.apache.log4j.Logger; + +import org.apache.qpid.framing.*; +import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9; +import org.apache.qpid.server.state.StateAwareMethodListener; +import org.apache.qpid.server.state.AMQStateManager; +import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.qpid.server.exchange.ExchangeRegistry; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.security.access.Permission; +import org.apache.qpid.AMQException; +import org.apache.qpid.AMQInvalidRoutingKeyException; +import org.apache.qpid.protocol.AMQConstant; + +public class QueueUnbindHandler implements StateAwareMethodListener<QueueUnbindBody> +{ + private static final Logger _log = Logger.getLogger(QueueUnbindHandler.class); + + private static final QueueUnbindHandler _instance = new QueueUnbindHandler(); + + public static QueueUnbindHandler getInstance() + { + return _instance; + } + + private QueueUnbindHandler() + { + } + + public void methodReceived(AMQStateManager stateManager, QueueUnbindBody body, int channelId) throws AMQException + { + AMQProtocolSession session = stateManager.getProtocolSession(); + VirtualHost virtualHost = session.getVirtualHost(); + ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); + QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); + + + final AMQQueue queue; + final AMQShortString routingKey; + + if (body.getQueue() == null) + { + AMQChannel channel = session.getChannel(channelId); + + if (channel == null) + { + throw body.getChannelNotFoundException(channelId); + } + + queue = channel.getDefaultQueue(); + + if (queue == null) + { + throw body.getConnectionException(AMQConstant.NOT_FOUND, "No default queue defined on channel and queue was null"); + } + + routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern(); + + } + else + { + queue = queueRegistry.getQueue(body.getQueue()); + routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern(); + } + + if (queue == null) + { + throw body.getConnectionException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist."); + } + final Exchange exch = exchangeRegistry.getExchange(body.getExchange()); + if (exch == null) + { + throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.getExchange() + " does not exist."); + } + + //Perform ACLs + virtualHost.getAccessManager().authorise(session, Permission.UNBIND, body, queue); + + try + { + queue.unBind(routingKey, body.getArguments(), exch); + } + catch (AMQInvalidRoutingKeyException rke) + { + throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString()); + } + catch (AMQException e) + { + if(e.getErrorCode() == AMQConstant.NOT_FOUND) + { + throw body.getConnectionException(AMQConstant.NOT_FOUND,e.getMessage(),e); + } + throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString()); + } + + if (_log.isInfoEnabled()) + { + _log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey); + } + + MethodRegistry_0_9 methodRegistry = (MethodRegistry_0_9) session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createQueueUnbindOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + + + } +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java index 4fb260472d..a0ecc2bd85 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java @@ -20,7 +20,7 @@ */ package org.apache.qpid.server.management; -import org.apache.qpid.server.security.access.UserManagement; +import org.apache.qpid.server.security.access.management.UserManagement; import org.apache.log4j.Logger; import javax.management.remote.MBeanServerForwarder; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java index f75b485c82..6aceea50c5 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java @@ -109,7 +109,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable private FieldTable _clientProperties; private final List<Task> _taskList = new CopyOnWriteArrayList<Task>(); - private List<Integer> _closingChannelsList = new ArrayList<Integer>(); + private List<Integer> _closingChannelsList = new CopyOnWriteArrayList<Integer>(); private ProtocolOutputConverter _protocolOutputConverter; private Principal _authorizedID; private MethodDispatcher _dispatcher; @@ -208,9 +208,39 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable { _logger.debug("Frame Received: " + frame); } + + // Check that this channel is not closing + if (channelAwaitingClosure(channelId)) + { + if ((frame.getBodyFrame() instanceof ChannelCloseOkBody)) + { + if (_logger.isInfoEnabled()) + { + _logger.info("Channel[" + channelId + "] awaiting closure - processing close-ok"); + } + } + else + { + if (_logger.isInfoEnabled()) + { + _logger.info("Channel[" + channelId + "] awaiting closure ignoring"); + } + + return; + } + } - body.handle(channelId, this); + + try + { + body.handle(channelId, this); + } + catch (AMQException e) + { + closeChannel(channelId); + throw e; + } } @@ -259,27 +289,6 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable final AMQMethodEvent<AMQMethodBody> evt = new AMQMethodEvent<AMQMethodBody>(channelId, methodBody); - // Check that this channel is not closing - if (channelAwaitingClosure(channelId)) - { - if ((evt.getMethod() instanceof ChannelCloseOkBody)) - { - if (_logger.isInfoEnabled()) - { - _logger.info("Channel[" + channelId + "] awaiting closure - processing close-ok"); - } - } - else - { - if (_logger.isInfoEnabled()) - { - _logger.info("Channel[" + channelId + "] awaiting closure ignoring"); - } - - return; - } - } - try { try @@ -341,6 +350,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable _logger.info("Closing connection due to: " + e.getMessage()); } + markChannelawaitingCloseOk(channelId); closeSession(); _stateManager.changeState(AMQState.CONNECTION_CLOSING); writeFrame(e.getCloseFrame(channelId)); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java index 45e0056d32..7696ca1e83 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java @@ -806,10 +806,6 @@ public class AMQMessage implements StorableMessage _transientMessageData = transientMessageData; } - public void clearTransientMessageData() - { - _transientMessageData = null; - } public String toString() { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java index b7cdaa29ab..882efd380d 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java @@ -271,8 +271,4 @@ class SubscriptionSet implements WeightedSubscriptionManager _exclusive = exclusive; } - public boolean getExcBoolean() - { - return _exclusive; - } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java index 42c32dcf00..e0fcaa208d 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java @@ -7,9 +7,9 @@ * 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 @@ -39,8 +39,8 @@ import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.database.ConfigurationFilePrincipalDatabaseManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager; -import org.apache.qpid.server.security.access.AccessManager; -import org.apache.qpid.server.security.access.AccessManagerImpl; +import org.apache.qpid.server.security.access.ACLPlugin; +import org.apache.qpid.server.security.access.ACLManager; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; import org.apache.qpid.AMQException; @@ -52,13 +52,14 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry private AuthenticationManager _authenticationManager; - private AccessManager _accessManager; + private ACLPlugin _accessManager; private PrincipalDatabaseManager _databaseManager; private VirtualHostRegistry _virtualHostRegistry; + //fixme Why is this not used. private final Map<String, VirtualHost> _virtualHosts = new ConcurrentHashMap<String, VirtualHost>(); private PluginManager _pluginManager; @@ -110,7 +111,7 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry _virtualHostRegistry = new VirtualHostRegistry(); - _accessManager = new AccessManagerImpl("default", _configuration); + _accessManager = ACLManager.loadACLManager("default", _configuration); _databaseManager = new ConfigurationFilePrincipalDatabaseManager(); @@ -121,7 +122,7 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry _managedObjectRegistry.start(); _pluginManager = new PluginManager(_configuration.getString("plugin-directory")); - + initialiseVirtualHosts(); } @@ -154,7 +155,7 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry return _virtualHostRegistry; } - public AccessManager getAccessManager() + public ACLPlugin getAccessManager() { return _accessManager; } @@ -178,7 +179,7 @@ public class ConfigurationFileApplicationRegistry extends ApplicationRegistry { return getConfiguration().getList("virtualhosts.virtualhost.name"); } - + public PluginManager getPluginManager() { return _pluginManager; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java index 6aac21a161..ca10fbdba2 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java @@ -27,7 +27,7 @@ import org.apache.qpid.server.management.ManagedObjectRegistry; import org.apache.qpid.server.plugins.PluginManager; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; -import org.apache.qpid.server.security.access.AccessManager; +import org.apache.qpid.server.security.access.ACLPlugin; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; public interface IApplicationRegistry @@ -68,8 +68,8 @@ public interface IApplicationRegistry VirtualHostRegistry getVirtualHostRegistry(); - AccessManager getAccessManager(); - + ACLPlugin getAccessManager(); + PluginManager getPluginManager(); } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessManagerImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLManager.java index 35d036d20f..539f32a732 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessManagerImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLManager.java @@ -23,33 +23,35 @@ package org.apache.qpid.server.security.access; import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.ConfigurationException; import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; +import org.apache.qpid.server.security.access.plugins.DenyAll; import org.apache.qpid.configuration.PropertyUtils; import org.apache.log4j.Logger; import java.util.List; import java.lang.reflect.Method; -import java.security.Principal; -public class AccessManagerImpl implements AccessManager +public class ACLManager { - private static final Logger _logger = Logger.getLogger(AccessManagerImpl.class); + private static final Logger _logger = Logger.getLogger(ACLManager.class); - AccessManager _accessManager; - - public AccessManagerImpl(String name, Configuration hostConfig) throws ConfigurationException + public static ACLPlugin loadACLManager(String name, Configuration hostConfig) throws ConfigurationException { + ACLPlugin aclPlugin = ApplicationRegistry.getInstance().getAccessManager(); + if (hostConfig == null) { - _logger.warn("No Configuration specified. Using default access controls for VirtualHost:'" + name + "'"); - return; + _logger.warn("No Configuration specified. Using default ACLPlugin '" + aclPlugin.getPluginName() + + "' for VirtualHost:'" + name + "'"); + return aclPlugin; } String accessClass = hostConfig.getString("security.access.class"); if (accessClass == null) { - _logger.warn("No access control specified. Using default access controls for VirtualHost:'" + name + "'"); - return; + + _logger.warn("No ACL Plugin specified. Using default ACL Plugin '" + aclPlugin.getPluginName() + + "' for VirtualHost:'" + name + "'"); + return aclPlugin; } Object o; @@ -59,26 +61,35 @@ public class AccessManagerImpl implements AccessManager } catch (Exception e) { - throw new ConfigurationException("Error initialising access control: " + e, e); + throw new ConfigurationException("Error initialising ACL: " + e, e); } - if (!(o instanceof AccessManager)) + if (!(o instanceof ACLPlugin)) { - throw new ConfigurationException("Access control must implement the VirtualHostAccess interface"); + throw new ConfigurationException("ACL Plugins must implement the ACLPlugin interface"); } - initialiseAccessControl((AccessManager) o, hostConfig); + initialiseAccessControl((ACLPlugin) o, hostConfig); - _accessManager = (AccessManager) o; - - _logger.info("Initialised access control for virtualhost '" + name + "' successfully"); + aclPlugin = getManager((ACLPlugin) o); + if (_logger.isInfoEnabled()) + { + _logger.info("Initialised ACL Plugin '" + aclPlugin.getPluginName() + + "' for virtualhost '" + name + "' successfully"); + } + return aclPlugin; } - private void initialiseAccessControl(AccessManager accessManager, Configuration config) + private static void initialiseAccessControl(ACLPlugin accessManager, Configuration config) throws ConfigurationException { + //First provide the ACLPlugin with the host configuration + + accessManager.setConfiguaration(config); + + //Provide additional attribute customisation. String baseName = "security.access.attributes.attribute."; List<String> argumentNames = config.getList(baseName + "name"); List<String> argumentValues = config.getList(baseName + "value"); @@ -123,33 +134,28 @@ public class AccessManagerImpl implements AccessManager } } - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return isAuthorized(accessObject, new UsernamePrincipal(username), AccessRights.Rights.READ); - } - public AccessResult isAuthorized(Accessable accessObject, Principal user, AccessRights.Rights rights) + private static ACLPlugin getManager(ACLPlugin manager) { - if (_accessManager == null) + if (manager == null) { - if (ApplicationRegistry.getInstance().getAccessManager() == this) + if (ApplicationRegistry.getInstance().getAccessManager() == null) { - _logger.warn("No Default access manager specified DENYING ALL ACCESS"); - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); + return new DenyAll(); } else { - return ApplicationRegistry.getInstance().getAccessManager().isAuthorized(accessObject, user, rights); + return ApplicationRegistry.getInstance().getAccessManager(); } } else { - return _accessManager.isAuthorized(accessObject, user, rights); + return manager; } } - public String getName() + public static Logger getLogger() { - return "AccessManagerImpl"; + return _logger; } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java new file mode 100644 index 0000000000..7855f147b4 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java @@ -0,0 +1,58 @@ +/* + * 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; + +import org.apache.qpid.framing.AMQMethodBody; + +import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.AMQConnectionException; +import org.apache.commons.configuration.Configuration; + + +public interface ACLPlugin +{ + /** + * Pseudo-Code: + * Identify requested RighConnectiont + * Lookup users ability for that right. + * if rightsExists + * Validate right on object + * Return result + * e.g + * User, CONSUME , Queue + * User, CONSUME , Exchange + RoutingKey + * User, PUBLISH , Exchange + RoutingKey + * User, CREATE , Exchange || Queue + * User, BIND , Exchange + RoutingKey + Queue + * + * @param session - The session requesting access + * @param permission - The permission requested + * @param parameters - The above objects that are used to authorise the request. + * @return The AccessResult decision + */ + //todo potential refactor this ConnectionException Out of here + AccessResult authorise(AMQProtocolSession session, Permission permission, AMQMethodBody body, Object... parameters) throws AMQConnectionException; + + String getPluginName(); + + void setConfiguaration(Configuration config); + +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java index b8d8fc605a..89cead69b3 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java @@ -30,15 +30,15 @@ public class AccessResult StringBuilder _authorizer; AccessStatus _status; - public AccessResult(AccessManager authorizer, AccessStatus status) + public AccessResult(ACLPlugin authorizer, AccessStatus status) { _status = status; - _authorizer = new StringBuilder(authorizer.getName()); + _authorizer = new StringBuilder(authorizer.getPluginName()); } - public void setAuthorizer(AccessManager authorizer) + public void setAuthorizer(ACLPlugin authorizer) { - _authorizer.append(authorizer.getName()); + _authorizer.append(authorizer.getPluginName()); } public String getAuthorizer() @@ -56,10 +56,10 @@ public class AccessResult return _status; } - public void addAuthorizer(AccessManager accessManager) + public void addAuthorizer(ACLPlugin accessManager) { _authorizer.insert(0, "->"); - _authorizer.insert(0, accessManager.getName()); + _authorizer.insert(0, accessManager.getPluginName()); } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AllowAll.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AllowAll.java deleted file mode 100644 index 1ddca3a64e..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AllowAll.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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; - -import java.security.Principal; - -public class AllowAll implements AccessManager -{ - - public AccessResult isAuthorized(Accessable accessObject, Principal username, AccessRights.Rights rights) - { - return new AccessResult(this, AccessResult.AccessStatus.GRANTED); - } - - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return new AccessResult(this, AccessResult.AccessStatus.GRANTED); - } - - public String getName() - { - return "AllowAll"; - } -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/DenyAll.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/DenyAll.java deleted file mode 100644 index bf40eeba4e..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/DenyAll.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * 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; - -import java.security.Principal; - -public class DenyAll implements AccessManager -{ - public AccessResult isAuthorized(Accessable accessObject, Principal username, AccessRights.Rights rights) - { - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); - } - - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); - } - - public String getName() - { - return "DenyAll"; - } -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/FileAccessManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/FileAccessManager.java deleted file mode 100644 index 291bc714ed..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/FileAccessManager.java +++ /dev/null @@ -1,183 +0,0 @@ -/* - * 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; - -import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; -import org.apache.log4j.Logger; - -import java.io.IOException; -import java.io.BufferedReader; -import java.io.FileReader; -import java.io.FileNotFoundException; -import java.io.File; -import java.util.regex.Pattern; -import java.security.Principal; - -/** - * Represents a user database where the account information is stored in a simple flat file. - * - * The file is expected to be in the form: username:password username1:password1 ... usernamen:passwordn - * - * where a carriage return separates each username/password pair. Passwords are assumed to be in plain text. - */ -public class FileAccessManager implements AccessManager -{ - private static final Logger _logger = Logger.getLogger(FileAccessManager.class); - - protected File _accessFile; - - protected Pattern _regexp = Pattern.compile(":"); - - private static final short USER_INDEX = 0; - private static final short VIRTUALHOST_INDEX = 1; - - public void setAccessFile(String accessFile) throws FileNotFoundException - { - File f = new File(accessFile); - _logger.info("FileAccessManager using file " + f.getAbsolutePath()); - _accessFile = f; - if (!f.exists()) - { - throw new FileNotFoundException("Cannot find access file " + f); - } - if (!f.canRead()) - { - throw new FileNotFoundException("Cannot read access file " + f + - ". Check permissions."); - } - } - - /** - * Looks up the virtual hosts for a specified user in the access file. - * - * @param user The user to lookup - * - * @return a list of virtualhosts - */ - private VirtualHostAccess[] lookupVirtualHost(String user) - { - String[] results = lookup(user, VIRTUALHOST_INDEX); - VirtualHostAccess vhosts[] = new VirtualHostAccess[results.length]; - - for (int index = 0; index < results.length; index++) - { - vhosts[index] = new VirtualHostAccess(results[index]); - } - - return vhosts; - } - - - private String[] lookup(String user, int index) - { - try - { - BufferedReader reader = null; - try - { - reader = new BufferedReader(new FileReader(_accessFile)); - String line; - - while ((line = reader.readLine()) != null) - { - String[] result = _regexp.split(line); - if (result == null || result.length < (index + 1)) - { - continue; - } - - if (user.equals(result[USER_INDEX])) - { - return result[index].split(","); - } - } - return null; - } - finally - { - if (reader != null) - { - reader.close(); - } - } - } - catch (IOException ioe) - { - //ignore - } - return null; - } - - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return isAuthorized(accessObject, new UsernamePrincipal(username), AccessRights.Rights.READ); - } - - public AccessResult isAuthorized(Accessable accessObject, Principal user, AccessRights.Rights rights) - { - if (accessObject instanceof VirtualHost) - { - VirtualHostAccess[] hosts = lookupVirtualHost(user.getName()); - - if (hosts != null) - { - for (VirtualHostAccess host : hosts) - { - if (accessObject.getAccessableName().equals(host.getVirtualHost())) - { - if (host.getAccessRights().allows(rights)) - { - return new AccessResult(this, AccessResult.AccessStatus.GRANTED); - } - else - { - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); - } - } - } - } - } -// else if (accessObject instanceof AMQQueue) -// { -// String[] queues = lookupQueue(username, ((AMQQueue) accessObject).getVirtualHost()); -// -// if (queues != null) -// { -// for (String queue : queues) -// { -// if (accessObject.getAccessableName().equals(queue)) -// { -// return new AccessResult(this, AccessResult.AccessStatus.GRANTED); -// } -// } -// } -// } - - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); - } - - public String getName() - { - return "FileAccessManager"; - } - -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/Permission.java index d70a6dc8f4..5d439a99eb 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessManager.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/Permission.java @@ -1,34 +1,37 @@ -/* - * 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; - -import java.security.Principal; - -public interface AccessManager -{ - AccessResult isAuthorized(Accessable accessObject, Principal username, AccessRights.Rights rights); - - @Deprecated - AccessResult isAuthorized(Accessable accessObject, String username); - - String getName(); - -} +/*
+ * 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;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.queue.AMQQueue;
+
+public enum Permission
+{
+ CONSUME,
+ PUBLISH,
+ CREATE,
+ ACCESS,
+ BIND,
+ UNBIND,
+ DELETE,
+ PURGE
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalDatabaseAccessManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalDatabaseAccessManager.java deleted file mode 100644 index 6ccadb2e7d..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalDatabaseAccessManager.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * 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; - -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.security.auth.database.PrincipalDatabase; -import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; -import org.apache.log4j.Logger; - -import java.security.Principal; - -public class PrincipalDatabaseAccessManager implements AccessManager -{ - private static final Logger _logger = Logger.getLogger(PrincipalDatabaseAccessManager.class); - - PrincipalDatabase _database; - AccessManager _default; - - public PrincipalDatabaseAccessManager() - { - _default = null; - } - - public void setDefaultAccessManager(String defaultAM) - { - if (defaultAM.equals("AllowAll")) - { - _default = new AllowAll(); - } - - if (defaultAM.equals("DenyAll")) - { - _default = new DenyAll(); - } - } - - public void setPrincipalDatabase(String database) - { - _database = ApplicationRegistry.getInstance().getDatabaseManager().getDatabases().get(database); - if (!(_database instanceof AccessManager)) - { - _logger.warn("Database '" + database + "' cannot perform access management"); - } - } - - - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return isAuthorized(accessObject, new UsernamePrincipal(username), AccessRights.Rights.READ); - } - - public AccessResult isAuthorized(Accessable accessObject, Principal username, AccessRights.Rights rights) - { - AccessResult result; - - if (_database == null) - { - if (_default != null) - { - result = _default.isAuthorized(accessObject, username, rights); - } - else - { - throw new RuntimeException("Principal Database and default Access Manager are both null unable to perform Access Control"); - } - } - else - { - if (!(_database instanceof AccessManager)) - { - _logger.warn("Specified PrincipalDatabase is not an AccessManager so using default AccessManager"); - result = _default.isAuthorized(accessObject, username, rights); - } - else - { - result = ((AccessManager) _database).isAuthorized(accessObject, username, rights); - } - } - - result.addAuthorizer(this); - - return result; - } - - public String getName() - { - return "PrincipalDatabaseFileAccessManager"; - } - -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java new file mode 100755 index 0000000000..22f1cf25a8 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java @@ -0,0 +1,587 @@ +/* + * + * 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; + +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.QueueBindBody; +import org.apache.qpid.framing.QueueDeclareBody; +import org.apache.qpid.framing.ExchangeDeclareBody; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.exchange.Exchange; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +public class PrincipalPermissions +{ + + private static final int CONSUME_QUEUES_KEY = 0; + private static final int CONSUME_TEMPORARY_KEY = 1; + private static final int CONSUME_OWN_QUEUES_ONLY_KEY = 2; + + private static final int CREATE_QUEUES_KEY = 0; + private static final int CREATE_EXCHANGES_KEY = 1; + + private static final int CREATE_QUEUE_TEMPORARY_KEY = 2; + private static final int CREATE_QUEUE_QUEUES_KEY = 1; + private static final int CREATE_QUEUE_EXCHANGES_KEY = 0; + + private static final int CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY = 0; + private static final int CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY = 1; + + private static final int PUBLISH_EXCHANGES_KEY = 0; + + private Map _permissions; + + private String _user; + + + public PrincipalPermissions(String user) + { + _user = user; + _permissions = new ConcurrentHashMap(); + } + + public void grant(Permission permission, Object... parameters) + { + switch (permission) + { + case ACCESS: + break; // This is a no-op as the existence of this PrincipalPermission object is scoped per VHost for ACCESS + case BIND: + break; // All the details are currently included in the create setup. + case CONSUME: // Parameters : AMQShortString queueName, Boolean Temporary, Boolean ownQueueOnly + Map consumeRights = (Map) _permissions.get(permission); + + if (consumeRights == null) + { + consumeRights = new ConcurrentHashMap(); + _permissions.put(permission, consumeRights); + } + + //if we have parametsre + if (parameters.length > 0) + { + AMQShortString queueName = (AMQShortString) parameters[0]; + Boolean temporary = (Boolean) parameters[1]; + Boolean ownQueueOnly = (Boolean) parameters[2]; + + if (temporary) + { + consumeRights.put(CONSUME_TEMPORARY_KEY, true); + } + else + { + consumeRights.put(CONSUME_TEMPORARY_KEY, false); + } + + if (ownQueueOnly) + { + consumeRights.put(CONSUME_OWN_QUEUES_ONLY_KEY, true); + } + else + { + consumeRights.put(CONSUME_OWN_QUEUES_ONLY_KEY, false); + } + + + LinkedList queues = (LinkedList) consumeRights.get(CONSUME_QUEUES_KEY); + if (queues == null) + { + queues = new LinkedList(); + consumeRights.put(CONSUME_QUEUES_KEY, queues); + } + + if (queueName != null) + { + queues.add(queueName); + } + } + + + break; + case CREATE: // Parameters : Boolean temporary, AMQShortString queueName + // , AMQShortString exchangeName , AMQShortString routingKey + // || AMQShortString exchangeName , AMQShortString Class + + Map createRights = (Map) _permissions.get(permission); + + if (createRights == null) + { + createRights = new ConcurrentHashMap(); + _permissions.put(permission, createRights); + + } + + //The existence of the empty map mean permission to all. + if (parameters.length == 0) + { + return; + } + + + if (parameters[0] instanceof Boolean) //Create Queue : + // Boolean temporary, [AMQShortString queueName, AMQShortString exchangeName , AMQShortString routingKey] + { + Boolean temporary = (Boolean) parameters[0]; + + AMQShortString queueName = parameters.length > 1 ? (AMQShortString) parameters[1] : null; + AMQShortString exchangeName = parameters.length > 2 ? (AMQShortString) parameters[2] : null; + //Set the routingkey to the specified value or the queueName if present + AMQShortString routingKey = parameters.length > 3 ? (AMQShortString) parameters[3] : queueName; + + // Get the queues map + Map create_queues = (Map) createRights.get(CREATE_QUEUES_KEY); + + if (create_queues == null) + { + create_queues = new ConcurrentHashMap(); + createRights.put(CREATE_QUEUES_KEY, create_queues); + } + + //Allow all temp queues to be created + create_queues.put(CREATE_QUEUE_TEMPORARY_KEY, temporary); + + //Create empty list of queues + Map create_queues_queues = (Map) create_queues.get(CREATE_QUEUE_QUEUES_KEY); + + if (create_queues_queues == null) + { + create_queues_queues = new ConcurrentHashMap(); + create_queues.put(CREATE_QUEUE_QUEUES_KEY, create_queues_queues); + } + + // We are granting CREATE rights to all temporary queues only + if (parameters.length == 1) + { + return; + } + + // if we have a queueName then we need to store any associated exchange / rk bindings + if (queueName != null) + { + Map queue = (Map) create_queues_queues.get(queueName); + if (queue == null) + { + queue = new ConcurrentHashMap(); + create_queues_queues.put(queueName, queue); + } + + if (exchangeName != null) + { + queue.put(exchangeName, routingKey); + } + + //If no exchange is specified then the presence of the queueName in the map says any exchange is ok + } + + // Store the exchange that we are being granted rights to. This will be used as part of binding + + //Lookup the list of exchanges + Map create_queues_exchanges = (Map) create_queues.get(CREATE_QUEUE_EXCHANGES_KEY); + + if (create_queues_exchanges == null) + { + create_queues_exchanges = new ConcurrentHashMap(); + create_queues.put(CREATE_QUEUE_EXCHANGES_KEY, create_queues_exchanges); + } + + //if we have an exchange + if (exchangeName != null) + { + //Retrieve the list of permitted exchanges. + Map exchanges = (Map) create_queues_exchanges.get(exchangeName); + + if (exchanges == null) + { + exchanges = new ConcurrentHashMap(); + create_queues_exchanges.put(exchangeName, exchanges); + } + + //Store the temporary setting CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY + exchanges.put(CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY, temporary); + + //Store the binding details of queue/rk for this exchange. + if (queueName != null) + { + //Retrieve the list of permitted routingKeys. + Map rKeys = (Map) exchanges.get(exchangeName); + + if (rKeys == null) + { + rKeys = new ConcurrentHashMap(); + exchanges.put(CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY, rKeys); + } + + rKeys.put(queueName, routingKey); + } + } + } + else // Create Exchange : AMQShortString exchangeName , AMQShortString Class + { + Map create_exchanges = (Map) createRights.get(CREATE_EXCHANGES_KEY); + + if (create_exchanges == null) + { + create_exchanges = new ConcurrentHashMap(); + createRights.put(CREATE_EXCHANGES_KEY, create_exchanges); + } + + //Should perhaps error if parameters[0] is null; + AMQShortString exchangeName = parameters.length > 0 ? (AMQShortString) parameters[0] : null; + AMQShortString className = parameters.length > 1 ? (AMQShortString) parameters[1] : null; + + //Store the exchangeName / class mapping if the mapping is null + createRights.put(exchangeName, className); + } + break; + case DELETE: + break; + + case PUBLISH: // Parameters : Exchange exchange, AMQShortString routingKey + Map publishRights = (Map) _permissions.get(permission); + + if (publishRights == null) + { + publishRights = new ConcurrentHashMap(); + _permissions.put(permission, publishRights); + } + + if (parameters == null || parameters.length == 0) + { + //If we have no parameters then allow publish to all destinations + // this is signified by having a null value for publish_exchanges + } + else + { + Map publish_exchanges = (Map) publishRights.get(PUBLISH_EXCHANGES_KEY); + + if (publish_exchanges == null) + { + publish_exchanges = new ConcurrentHashMap(); + publishRights.put(PUBLISH_EXCHANGES_KEY, publish_exchanges); + } + + + HashSet routingKeys = (HashSet) publish_exchanges.get(parameters[0]); + + // Check to see if we have a routing key + if (parameters.length == 2) + { + if (routingKeys == null) + { + routingKeys = new HashSet<AMQShortString>(); + } + //Add routing key to permitted publish destinations + routingKeys.add(parameters[1]); + } + + // Add the updated routingkey list or null if all values allowed + publish_exchanges.put(parameters[0], routingKeys); + } + break; + case PURGE: + break; + case UNBIND: + break; + } + + } + + public boolean authorise(Permission permission, Object... parameters) + { + + switch (permission) + { + case ACCESS: + return true; // This is here for completeness but the SimpleXML ACLManager never calls it. + // The existence of this user specific PP can be validated in the map SimpleXML maintains. + case BIND: // Parameters : QueueBindMethod , Exchange , AMQQueue, AMQShortString routingKey + +// QueueDeclareBody body = (QueueDeclareBody) parameters[0]; + + Exchange exchange = (Exchange) parameters[1]; + + if (exchange.getName().equals("<<default>>")) + { + // Binding to <<default>> can not be programmed via ACLs due to '<','>' unable to be used in the XML + System.err.println("Binding on exchange <<default>> not alowed via ACLs"); + } + + AMQQueue bind_queueName = (AMQQueue) parameters[2]; + AMQShortString routingKey = (AMQShortString) parameters[3]; + + //Get all Create Rights for this user + Map bindCreateRights = (Map) _permissions.get(Permission.CREATE); + + //Look up the Queue Creation Rights + Map bind_create_queues = (Map) bindCreateRights.get(CREATE_QUEUES_KEY); + + //Lookup the list of queues + Map bind_create_queues_queues = (Map) bindCreateRights.get(CREATE_QUEUE_QUEUES_KEY); + + // Check and see if we have a queue white list to check + if (bind_create_queues_queues != null) + { + //There a white list for queues + Map exchangeDetails = (Map) bind_create_queues_queues.get(bind_queueName); + + if (exchangeDetails == null) //Then all queue can be bound to all exchanges. + { + return true; + } + + // Check to see if we have a white list of routingkeys to check + Map rkeys = (Map) exchangeDetails.get(exchange.getName()); + + // if keys is null then any rkey is allowed on this exchange + if (rkeys == null) + { + // There is no routingkey white list + return true; + } + else + { + // We have routingKeys so a match must be found to allowed binding + Iterator keys = rkeys.keySet().iterator(); + + boolean matched = false; + while (keys.hasNext() && !matched) + { + AMQShortString rkey = (AMQShortString) keys.next(); + if (rkey.endsWith("*")) + { + matched = routingKey.startsWith(rkey.subSequence(0, rkey.length() - 1).toString()); + } + else + { + matched = routingKey.equals(rkey); + } + } + + + return matched; + } + + + } + else + { + //There a is no white list for queues + + // So can allow all queues to be bound + // but we should first check and see if we have a temp queue and validate that we are allowed + // to bind temp queues. + + //Check to see if we have a temporary queue + if (bind_queueName.isAutoDelete()) + { + // Check and see if we have an exchange white list. + Map bind_exchanges = (Map) bind_create_queues.get(CREATE_QUEUE_EXCHANGES_KEY); + + // If the exchange exists then we must check to see if temporary queues are allowed here + if (bind_exchanges != null) + { + // Check to see if the requested exchange is allowed. + Map exchangeDetails = (Map) bind_exchanges.get(exchange.getName()); + + return (Boolean) exchangeDetails.get(CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY); + } + + //no white list so all allowed, drop through to return true below. + } + + // not a temporary queue and no white list so all allowed. + return true; + } + + case CREATE:// Paramters : QueueDeclareBody || ExchangeDeclareBody + + Map createRights = (Map) _permissions.get(permission); + + // If there are no create rights then deny request + if (createRights == null) + { + return false; + } + + if (parameters.length == 1) + { + if (parameters[0] instanceof QueueDeclareBody) + { + QueueDeclareBody body = (QueueDeclareBody) parameters[0]; + + //Look up the Queue Creation Rights + Map create_queues = (Map) createRights.get(CREATE_QUEUES_KEY); + + //Lookup the list of queues allowed to be created + Map create_queues_queues = (Map) create_queues.get(CREATE_QUEUE_QUEUES_KEY); + + + AMQShortString queueName = body.getQueue(); + + + if (body.getAutoDelete())// we have a temporary queue + { + return (Boolean) create_queues.get(CREATE_QUEUE_TEMPORARY_KEY); + } + else + { + // If there is a white list then check + return create_queues_queues == null || create_queues_queues.containsKey(queueName); + } + + } + else if (parameters[0] instanceof ExchangeDeclareBody) + { + ExchangeDeclareBody body = (ExchangeDeclareBody) parameters[0]; + + AMQShortString exchangeName = body.getExchange(); + + Map create_exchanges = (Map) createRights.get(CREATE_EXCHANGES_KEY); + + // If the exchange list is doesn't exist then all is allowed else check the valid exchanges + return create_exchanges == null || create_exchanges.containsKey(exchangeName); + } + } + break; + case CONSUME: // Parameters : AMQQueue + + if (parameters.length == 1 && parameters[0] instanceof AMQQueue) + { + AMQQueue queue = ((AMQQueue) parameters[0]); + Map queuePermissions = (Map) _permissions.get(permission); + + List queues = (List) queuePermissions.get(CONSUME_QUEUES_KEY); + + Boolean temporayQueues = (Boolean) queuePermissions.get(CONSUME_TEMPORARY_KEY); + Boolean ownQueuesOnly = (Boolean) queuePermissions.get(CONSUME_OWN_QUEUES_ONLY_KEY); + + // If user is allowed to publish to temporary queues and this is a temp queue then allow it. + if (temporayQueues) + { + if (queue.isAutoDelete()) + // This will allow consumption from any temporary queue including ones not owned by this user. + // Of course the exclusivity will not be broken. + { + // if not limited to ownQueuesOnly then ok else check queue Owner. + return !ownQueuesOnly || queue.getOwner().equals(_user); + } + else + { + return false; + } + } + + // if queues are white listed then ensure it is ok + if (queues != null) + { + // if no queues are listed then ALL are ok othereise it must be specified. + if (ownQueuesOnly) + { + if (queue.getOwner().equals(_user)) + { + return queues.size() == 0 || queues.contains(queue.getName()); + } + else + { + return false; + } + } + + // If we are + return queues.size() == 0 || queues.contains(queue.getName()); + } + } + + // Can't authenticate without the right parameters + return false; + case DELETE: + break; + + case PUBLISH: // Parameters : Exchange exchange, AMQShortString routingKey + Map publishRights = (Map) _permissions.get(permission); + + if (publishRights == null) + { + return false; + } + + Map exchanges = (Map) publishRights.get(PUBLISH_EXCHANGES_KEY); + + // Having no exchanges listed gives full publish rights to all exchanges + if (exchanges == null) + { + return true; + } + // Otherwise exchange must be listed in the white list + + // If the map doesn't have the exchange then it isn't allowed + if (!exchanges.containsKey(parameters[0])) + { + return false; + } + else + { + + // Get valid routing keys + HashSet routingKeys = (HashSet) exchanges.get(parameters[0]); + + // Having no routingKeys in the map then all are allowed. + if (routingKeys == null) + { + return true; + } + else + { + // We have routingKeys so a match must be found to allowed binding + Iterator keys = routingKeys.iterator(); + + + AMQShortString publishRKey = (AMQShortString)parameters[1]; + + boolean matched = false; + while (keys.hasNext() && !matched) + { + AMQShortString rkey = (AMQShortString) keys.next(); + + if (rkey.endsWith("*")) + { + matched = publishRKey.startsWith(rkey.subSequence(0, rkey.length() - 1)); + } + else + { + matched = publishRKey.equals(rkey); + } + } + return matched; + } + } + case PURGE: + break; + case UNBIND: + break; + + } + + return false; + } +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AMQUserManagementMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/management/AMQUserManagementMBean.java index 2dc7fcbc1e..a8ae03cc5d 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/AMQUserManagementMBean.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/management/AMQUserManagementMBean.java @@ -18,7 +18,7 @@ * * */ -package org.apache.qpid.server.security.access; +package org.apache.qpid.server.security.access.management; import org.apache.qpid.server.management.MBeanDescription; import org.apache.qpid.server.management.AMQManagedObject; @@ -26,6 +26,7 @@ import org.apache.qpid.server.management.MBeanOperation; import org.apache.qpid.server.management.MBeanInvocationHandlerImpl; import org.apache.qpid.server.security.auth.database.PrincipalDatabase; import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; +import org.apache.qpid.server.security.access.management.UserManagement; import org.apache.log4j.Logger; import org.apache.commons.configuration.ConfigurationException; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/UserManagement.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java index b8762aa43b..658d7ebbd3 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/UserManagement.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java @@ -18,7 +18,7 @@ * * */ -package org.apache.qpid.server.security.access; +package org.apache.qpid.server.security.access.management; import org.apache.qpid.server.management.MBeanOperation; import org.apache.qpid.server.management.MBeanOperationParameter; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java new file mode 100644 index 0000000000..a51061aa0d --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java @@ -0,0 +1,68 @@ +/* + * 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.qpid.framing.AMQMethodBody; +import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.security.access.ACLPlugin; +import org.apache.qpid.server.security.access.ACLManager; +import org.apache.qpid.server.security.access.AccessResult; +import org.apache.qpid.server.security.access.Accessable; +import org.apache.qpid.server.security.access.Permission; +import org.apache.commons.configuration.Configuration; + +public class AllowAll implements ACLPlugin +{ + public AccessResult authorise(AMQProtocolSession session, Permission permission, AMQMethodBody body, Object... parameters) + { + if (ACLManager.getLogger().isInfoEnabled()) + { + ACLManager.getLogger().info("Allowing user:" + session.getAuthorizedID() + " for :" + permission.toString() + + " on " + body.getClass().getSimpleName() + + (parameters == null || parameters.length == 0 ? "" : "-" + accessablesToString(parameters))); + } + + return new AccessResult(this, AccessResult.AccessStatus.GRANTED); + } + + public static String accessablesToString(Object[] accessObject) + { + StringBuilder sb = new StringBuilder(); + + for (Object access : accessObject) + { + sb.append(access.getClass().getSimpleName() + ":" + access.toString() + ", "); + } + + return sb.delete(sb.length() - 2, sb.length()).toString(); + } + + public String getPluginName() + { + return "AllowAll"; + } + + public void setConfiguaration(Configuration config) + { + //no-op + } + +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java new file mode 100644 index 0000000000..80c125e737 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java @@ -0,0 +1,57 @@ +/* + * 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.qpid.framing.AMQMethodBody; +import org.apache.qpid.protocol.AMQConstant; +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.AMQConnectionException; +import org.apache.commons.configuration.Configuration; + +public class DenyAll implements ACLPlugin +{ + public AccessResult authorise(AMQProtocolSession session, Permission permission, AMQMethodBody body, Object... parameters) throws AMQConnectionException + { + + if (ACLManager.getLogger().isInfoEnabled()) + { + } + ACLManager.getLogger().info("Denying user:" + session.getAuthorizedID() + " for :" + permission.toString() + + " on " + body.getClass().getSimpleName() + + (parameters == null || parameters.length == 0 ? "" : "-" + AllowAll.accessablesToString(parameters))); + + throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, "DenyAll Plugin"); + } + + public String getPluginName() + { + return "DenyAll"; + } + + public void setConfiguaration(Configuration config) + { + //no-op + } +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java new file mode 100644 index 0000000000..c09cdc33f5 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java @@ -0,0 +1,431 @@ +/* + * 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 static final Logger _logger = ACLManager.getLogger(); + + private Map<String, PrincipalPermissions> _users; + + public SimpleXML() + { + _users = new ConcurrentHashMap<String, PrincipalPermissions>(); + } + + public void setConfiguaration(Configuration config) + { + _logger.info("SimpleXML Configuration"); + + 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); + _logger.info("PUBLISH:GRANTED:USER:" + user + " for all destinations"); + } + + // 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); + _logger.info("PUBLISH:GRANTED:USER:" + user + " on Exchange '" + exchangeName + "' for key '" + 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); + _logger.info("PUBLISH:GRANTED:USER:" + user + " on Exchange:" + 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); + if (temporary) + { + if (ownQueues) + { + _logger.info("CONSUME:GRANTED:USER:" + user + " on temporary queues owned by user."); + } + else + { + _logger.info("CONSUME:GRANTED:USER:" + user + " on all temporary queues."); + } + } + else + { + _logger.info("CONSUME:GRANTED:USER:" + user + " on queue '" + queueName + "'"); + } + } + + //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); + _logger.info("CONSUME:GRANTED:USER:" + user + " from all queues."); + } + } + + 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)); + + _logger.info("CREATE :GRANTED:USER:" + user + " for " + + (queueName.equals("") ? "" : "queue '" + queueName + "' ") + + (exchange.equals("") ? "" : "exchange '" + exchange + "' ") + + (routingKey.equals("") ? "" : " rk '" + routingKey + "' ") + + (temporary ? " temporary:" + temporary : "")); + } + + //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); + if (temporary) + { + _logger.info("CREATE :GRANTED:USER:" + user + " from temporary queues on any exchange."); + } + else + { + _logger.info("CREATE :GRANTED:USER:" + user + " from queue '" + queueName + "' on any exchange."); + } + } + + //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); + _logger.info("CREATE:GRANTED:USER:" + user + " for exchange '" + exchange + ":class:'" + 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); + _logger.info("CREATE:GRANTED:USER:" + user + " from all queues & exchanges."); + } + + + } + + 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); + + _logger.warn("Processing :" + permission + " for:" + username + ":" + permissions+":"+parameters.length); + + if (permissions != null) + { + switch (permission) + { + case ACCESS: + _logger.warn("GRANTED:"+permission); + return new AccessResult(this, AccessResult.AccessStatus.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])) + { + _logger.warn("GRANTED:"+permission); + return new AccessResult(this, AccessResult.AccessStatus.GRANTED); + } + } + break; + case CONSUME: // Parameters : none + if (parameters.length == 1 && permissions.authorise(Permission.CONSUME, parameters[0])) + { + _logger.warn("GRANTED:"+permission); + return new AccessResult(this, AccessResult.AccessStatus.GRANTED); + } + break; + case CREATE: // Body : QueueDeclareBody | ExchangeDeclareBody - Parameters : none + if (permissions.authorise(Permission.CREATE, body)) + { + _logger.warn("GRANTED:"+permission); + return new AccessResult(this, AccessResult.AccessStatus.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())) + { + _logger.warn("GRANTED:"+permission); + return new AccessResult(this, AccessResult.AccessStatus.GRANTED); + } + } + break; + case PURGE: + break; + case DELETE: + break; + case UNBIND: + break; + } + } + + _logger.warn("Access Denied for :" + permission + " for:" + username + ":" + permissions); + //todo potential refactor this ConnectionException Out of here + throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, error); + } + +//todo use or lose +// if (accessObject instanceof VirtualHost) +// { +// VirtualHostAccess[] hosts = lookupVirtualHost(user.getName()); +// +// if (hosts != null) +// { +// for (VirtualHostAccess host : hosts) +// { +// if (accessObject.getAccessableName().equals(host.getVirtualHost())) +// { +// if (host.getAccessRights().allows(rights)) +// { +// return new AccessResult(this, AccessResult.AccessStatus.GRANTED); +// } +// else +// { +// return new AccessResult(this, AccessResult.AccessStatus.REFUSED); +// } +// } +// } +// } +// } +// else if (accessObject instanceof AMQQueue) +// { +// String[] queues = lookupQueue(username, ((AMQQueue) accessObject).getVirtualHost()); +// +// if (queues != null) +// { +// for (String queue : queues) +// { +// if (accessObject.getAccessableName().equals(queue)) +// { +// return new AccessResult(this, AccessResult.AccessStatus.GRANTED); +// } +// } +// } +// } + +// return new AccessResult(this, AccessResult.AccessStatus.REFUSED); +// } + +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java index 10adfdd9fc..348bccb4e9 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java @@ -24,7 +24,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser; import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5HashedInitialiser; -import org.apache.qpid.server.security.access.AMQUserManagementMBean; +import org.apache.qpid.server.security.access.management.AMQUserManagementMBean; import org.apache.commons.codec.binary.Base64; import org.apache.commons.codec.EncoderException; @@ -45,7 +45,6 @@ import java.util.LinkedList; import java.util.concurrent.locks.ReentrantLock; import java.security.Principal; import java.security.NoSuchAlgorithmException; -import java.security.MessageDigest; /** * Represents a user database where the account information is stored in a simple flat file. diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java index 06eb9329a6..29055d9523 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java @@ -41,6 +41,8 @@ import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.security.access.AMQUserManagementMBean; import org.apache.qpid.server.security.auth.database.PrincipalDatabase; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; +import org.apache.qpid.server.security.access.management.AMQUserManagementMBean; +import org.apache.qpid.AMQException; public class ConfigurationFilePrincipalDatabaseManager implements PrincipalDatabaseManager { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordVhostFilePrincipalDatabase.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordVhostFilePrincipalDatabase.java deleted file mode 100644 index 5c372f6c2c..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordVhostFilePrincipalDatabase.java +++ /dev/null @@ -1,130 +0,0 @@ -/* - * 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.auth.database; - -import org.apache.log4j.Logger; -import org.apache.qpid.server.security.access.AccessManager; -import org.apache.qpid.server.security.access.AccessResult; -import org.apache.qpid.server.security.access.AccessRights; -import org.apache.qpid.server.security.access.Accessable; -import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal; -import org.apache.qpid.server.virtualhost.VirtualHost; - -import java.io.BufferedReader; -import java.io.FileReader; -import java.io.IOException; -import java.security.Principal; - -/** - * Represents a user database where the account information is stored in a simple flat file. - * - * The file is expected to be in the form: username:password username1:password1 ... usernamen:passwordn - * - * where a carriage return separates each username/password pair. Passwords are assumed to be in plain text. - */ -public class PlainPasswordVhostFilePrincipalDatabase extends PlainPasswordFilePrincipalDatabase implements AccessManager -{ - private static final Logger _logger = Logger.getLogger(PlainPasswordVhostFilePrincipalDatabase.class); - - /** - * Looks up the virtual hosts for a specified user in the password file. - * - * @param user The user to lookup - * - * @return a list of virtualhosts - */ - private String[] lookupVirtualHost(String user) - { - try - { - BufferedReader reader = null; - try - { - reader = new BufferedReader(new FileReader(_passwordFile)); - String line; - - while ((line = reader.readLine()) != null) - { - if (!line.startsWith("#")) - { - String[] result = _regexp.split(line); - if (result == null || result.length < 3) - { - continue; - } - - if (user.equals(result[0])) - { - return result[2].split(","); - } - } - } - return null; - } - finally - { - if (reader != null) - { - reader.close(); - } - } - } - catch (IOException ioe) - { - //ignore - } - return null; - } - - - public AccessResult isAuthorized(Accessable accessObject, String username) - { - return isAuthorized(accessObject, new UsernamePrincipal(username), AccessRights.Rights.READ); - } - - public AccessResult isAuthorized(Accessable accessObject, Principal user, AccessRights.Rights rights) - { - - if (accessObject instanceof VirtualHost) - { - String[] hosts = lookupVirtualHost(user.getName()); - - if (hosts != null) - { - for (String host : hosts) - { - if (accessObject.getAccessableName().equals(host)) - { - return new AccessResult(this, AccessResult.AccessStatus.GRANTED); - } - } - } - } - - return new AccessResult(this, AccessResult.AccessStatus.REFUSED); - } - - public String getName() - { - return "PlainPasswordVhostFile"; - } - -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java index ce5e0cd748..f589140e8e 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java @@ -82,14 +82,6 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan if (databaseName == null) { - if (hostConfig instanceof SubsetConfiguration) - { - _logger.warn("No authentication specified for '" + ((SubsetConfiguration) hostConfig).getPrefix() + "'. Using Default authentication manager"); - } - else - { - _logger.warn("No authentication specified. Using Default authentication manager"); - } _default = ApplicationRegistry.getInstance().getAuthenticationManager(); return; } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java index 102c97c458..5d9b3c42c7 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.java @@ -35,8 +35,8 @@ import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; import org.apache.qpid.server.security.auth.database.PropertiesPrincipalDatabaseManager; -import org.apache.qpid.server.security.access.AccessManager; -import org.apache.qpid.server.security.access.AllowAll; +import org.apache.qpid.server.security.access.ACLPlugin; +import org.apache.qpid.server.security.access.plugins.AllowAll; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; @@ -48,7 +48,7 @@ public class NullApplicationRegistry extends ApplicationRegistry private VirtualHostRegistry _virtualHostRegistry; - private AccessManager _accessManager; + private ACLPlugin _accessManager; private PrincipalDatabaseManager _databaseManager; @@ -121,7 +121,7 @@ public class NullApplicationRegistry extends ApplicationRegistry return _virtualHostRegistry; } - public AccessManager getAccessManager() + public ACLPlugin getAccessManager() { return _accessManager; } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index 06d70768fc..e68c520f40 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java @@ -26,8 +26,8 @@ import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.PropertiesConfiguration; import org.apache.log4j.Logger; import org.apache.qpid.server.AMQBrokerManagerMBean; -import org.apache.qpid.server.security.access.AccessManager; -import org.apache.qpid.server.security.access.AccessManagerImpl; +import org.apache.qpid.server.security.access.ACLPlugin; +import org.apache.qpid.server.security.access.ACLManager; import org.apache.qpid.server.security.access.Accessable; import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; @@ -72,7 +72,7 @@ public class VirtualHost implements Accessable private AuthenticationManager _authenticationManager; - private AccessManager _accessManager; + private ACLPlugin _accessManager; private final Timer _houseKeepingTimer = new Timer("Queue-housekeeping", true); @@ -172,7 +172,7 @@ public class VirtualHost implements Accessable _authenticationManager = new PrincipalDatabaseAuthenticationManager(name, hostConfig); - _accessManager = new AccessManagerImpl(name, hostConfig); + _accessManager = ACLManager.loadACLManager(name, hostConfig); _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean); _brokerMBean.register(); @@ -302,7 +302,7 @@ public class VirtualHost implements Accessable return _authenticationManager; } - public AccessManager getAccessManager() + public ACLPlugin getAccessManager() { return _accessManager; } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index 18d4625596..80bed1a131 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -470,7 +470,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect if (exceptions.size() > 0) { - JMSException e = exceptions.get(exceptions.size() - 1); + JMSException e = exceptions.get(0); int code = -1; try { @@ -875,11 +875,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect { long startCloseTime = System.currentTimeMillis(); - _taskPool.shutdown(); - closeAllSessions(null, timeout, startCloseTime); + closeAllSessions(null, timeout, startCloseTime); - if (!_taskPool.isTerminated()) - { + //This MUST occur after we have successfully closed all Channels/Sessions + _taskPool.shutdown(); + + if (!_taskPool.isTerminated()) + { try { // adjust timeout diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index 95d9b45b10..9baccdb9d9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -639,7 +639,6 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess final AMQFrame frame = body.generateFrame(getChannelId()); getProtocolHandler().syncWrite(frame, ChannelCloseOkBody.class, timeout); - // When control resumes at this point, a reply will have been received that // indicates the broker has closed the channel successfully. @@ -1737,11 +1736,6 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess { JMSException ex = new JMSException("Error registering consumer: " + e); - if (_logger.isDebugEnabled()) - { - e.printStackTrace(); - } - ex.setLinkedException(e); throw ex; } @@ -2354,7 +2348,7 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess * * @todo Be aware of possible changes to parameter order as versions change. */ - private AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler) + public AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler) throws AMQException { /*return new FailoverRetrySupport<AMQShortString, AMQException>(*/ @@ -2427,17 +2421,17 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess return ++_nextProducerId; } - protected AMQProtocolHandler getProtocolHandler() + public AMQProtocolHandler getProtocolHandler() { return _connection.getProtocolHandler(); } - protected byte getProtocolMajorVersion() + public byte getProtocolMajorVersion() { return getProtocolHandler().getProtocolMajorVersion(); } - protected byte getProtocolMinorVersion() + public byte getProtocolMinorVersion() { return getProtocolHandler().getProtocolMinorVersion(); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index 8039e3a163..dd129fdbfa 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -632,7 +632,7 @@ public class AMQSession_0_10 extends AMQSession } } - protected AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler) + public AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler) throws AMQException { /*return new FailoverRetrySupport<AMQShortString, AMQException>(*/ diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java index fb2d72267b..7c211975d7 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java @@ -141,6 +141,9 @@ public class FailoverHandler implements Runnable _logger.info("Failover process veto-ed by client"); _amqProtocolHandler.setStateManager(existingStateManager); + + //todo: ritchiem these exceptions are useless... Would be better to attempt to propogate exception that + // prompted the failover event. if (_host != null) { _amqProtocolHandler.getConnection().exceptionReceived(new AMQDisconnectedException("Redirect was vetoed by client", null)); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java index cea058b604..950a3288fc 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java @@ -73,10 +73,11 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener<Co if (errorCode != AMQConstant.REPLY_SUCCESS) { - if (errorCode == AMQConstant.NOT_ALLOWED) + if (errorCode == AMQConstant.NOT_ALLOWED || (errorCode == AMQConstant.ACCESS_REFUSED)) { - _logger.info("Authentication Error:" + Thread.currentThread().getName()); + _logger.info("Error :" + errorCode +":"+ Thread.currentThread().getName()); + // todo ritchiem : Why do this here when it is going to be done in the finally block? session.closeProtocolSession(); // todo this is a bit of a fudge (could be conssidered such as each new connection needs a new state manager or at least a fresh state. @@ -98,6 +99,8 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener<Co session.closeProtocolSession(); + // ritchiem: Doing this though will cause any waiting connection start to be released without being able to + // see what the cause was. stateManager.changeState(AMQState.CONNECTION_CLOSED); } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index 7cda410f97..a158a681b9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -209,6 +209,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter } catch (RuntimeException e) { + _logger.warn(e.getMessage()); e.printStackTrace(); } @@ -352,18 +353,27 @@ public class AMQProtocolHandler extends IoHandlerAdapter */ public void exceptionCaught(IoSession session, Throwable cause) { - if (_failoverState == FailoverState.NOT_STARTED) - { - // if (!(cause instanceof AMQUndeliveredException) && (!(cause instanceof AMQAuthenticationException))) - if (cause instanceof AMQConnectionClosedException) - { - _logger.info("Exception caught therefore going to attempt failover: " + cause, cause); - // this will attemp failover - - sessionClosed(session); - _connection.exceptionReceived(cause); - } - + if (_failoverState == FailoverState.NOT_STARTED) + { + // if (!(cause instanceof AMQUndeliveredException) && (!(cause instanceof AMQAuthenticationException))) + if (cause instanceof AMQConnectionClosedException) + { + _logger.info("Exception caught therefore going to attempt failover: " + cause, cause); + // this will attemp failover + + sessionClosed(session); + _connection.exceptionReceived(cause); + } + + if (cause instanceof ProtocolCodecException) + { + _logger.info("Protocol Exception caught NOT going to attempt failover as " + + "cause isn't AMQConnectionClosedException: " + cause, cause); + + AMQException amqe = new AMQException(null, "Protocol handler error: " + cause, cause); + propagateExceptionToWaiters(amqe); + } + _connection.exceptionReceived(cause); // FIXME Need to correctly handle other exceptions. Things like ... // if (cause instanceof AMQChannelClosedException) // which will cause the JMSSession to end due to a channel close and so that Session needs diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java index 35f51261be..795a9c24b6 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java @@ -76,7 +76,7 @@ public class TransportConnection return _openSocketRegister.remove(socketID); } - public static ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException + public static synchronized ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException { int transport = getTransport(details.getTransport()); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java index b91fc2d960..b830c377b8 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java @@ -50,4 +50,8 @@ public interface MessageProducer extends javax.jms.MessageProducer void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, boolean mandatory, boolean immediate) throws JMSException; + + void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, + boolean mandatory, boolean immediate, boolean waitUntilSent) throws JMSException; + } diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java index a246352d8b..2fe01fc126 100644 --- a/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java @@ -183,7 +183,7 @@ public class TestLargePublisher } catch (UnknownHostException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } catch (AMQException e) { diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java b/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java index 33891142b5..37b4ff1498 100644 --- a/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java @@ -133,7 +133,7 @@ public class TestPublisher } catch (JMSException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } } @@ -163,7 +163,7 @@ public class TestPublisher } catch (UnknownHostException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } catch (AMQException e) { diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java index 7103397ad4..c97e2c4cb1 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java @@ -33,6 +33,7 @@ import org.apache.qpid.jms.Session; import junit.framework.TestCase; import javax.jms.Connection; +import javax.jms.JMSException; import javax.jms.QueueSession; import javax.jms.TopicSession; @@ -115,8 +116,8 @@ public class ConnectionTest extends TestCase } } - //fixme AMQAuthenticationException is not propogaged - public void PasswordFailureConnection() throws Exception + //See QPID-771 + public void testPasswordFailureConnection() throws Exception { try { @@ -125,10 +126,9 @@ public class ConnectionTest extends TestCase } catch (AMQException amqe) { - if (!(amqe instanceof AMQAuthenticationException)) - { - fail("Correct exception not thrown. Excpected 'AMQAuthenticationException' got: " + amqe); - } + assertEquals("Exception was wrong type", JMSException.class, amqe.getCause().getClass()); + Exception linked = ((JMSException) amqe.getCause()).getLinkedException(); + assertEquals("Exception was wrong type", AMQAuthenticationException.class, linked.getClass()); } } diff --git a/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java b/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java index 09cf785108..c515263317 100644 --- a/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java +++ b/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java @@ -1,34 +1,3 @@ -package org.apache.mina.common.support; - -import org.apache.mina.common.IoFuture; -import org.apache.mina.common.IoSession; -import org.apache.mina.common.IoFutureListener; - -import java.util.List; -import java.util.ArrayList; -import java.util.Iterator; - -/* -* -* 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. -* -*/ - /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -48,6 +17,15 @@ import java.util.Iterator; * under the License. * */ +package org.apache.mina.common.support; + +import org.apache.mina.common.IoFuture; +import org.apache.mina.common.IoSession; +import org.apache.mina.common.IoFutureListener; + +import java.util.List; +import java.util.ArrayList; +import java.util.Iterator; /** * A default implementation of {@link org.apache.mina.common.IoFuture}. diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java index fcd336b180..fe04155bb8 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java @@ -36,8 +36,5 @@ public interface AMQBody public void writePayload(ByteBuffer buffer); - //public void populateFromBuffer(ByteBuffer buffer, long size) - // throws AMQFrameDecodingException, AMQProtocolVersionException; - void handle(final int channelId, final AMQVersionAwareProtocolSession amqMinaProtocolSession) throws AMQException; } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java index ce0b92293d..505c819bb2 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java @@ -1,627 +1,692 @@ -/*
- *
- * 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.framing;
-
-import org.apache.mina.common.ByteBuffer;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.*;
-import java.lang.ref.WeakReference;
-
-/**
- * A short string is a representation of an AMQ Short String
- * Short strings differ from the Java String class by being limited to on ASCII characters (0-127)
- * and thus can be held more effectively in a byte buffer.
- *
- */
-public final class AMQShortString implements CharSequence, Comparable<AMQShortString>
-{
- private static final byte MINUS = (byte)'-';
- private static final byte ZERO = (byte) '0';
-
-
-
- private final class TokenizerImpl implements AMQShortStringTokenizer
- {
- private final byte _delim;
- private int _count = -1;
- private int _pos = 0;
-
- public TokenizerImpl(final byte delim)
- {
- _delim = delim;
- }
-
- public int countTokens()
- {
- if(_count == -1)
- {
- _count = 1 + AMQShortString.this.occurences(_delim);
- }
- return _count;
- }
-
- public AMQShortString nextToken()
- {
- if(_pos <= AMQShortString.this.length())
- {
- int nextDelim = AMQShortString.this.indexOf(_delim, _pos);
- if(nextDelim == -1)
- {
- nextDelim = AMQShortString.this.length();
- }
-
- AMQShortString nextToken = AMQShortString.this.substring(_pos, nextDelim++);
- _pos = nextDelim;
- return nextToken;
- }
- else
- {
- return null;
- }
- }
-
- public boolean hasMoreTokens()
- {
- return _pos <= AMQShortString.this.length();
- }
- }
-
- private AMQShortString substring(final int from, final int to)
- {
- return new AMQShortString(_data, from, to);
- }
-
-
- private static final ThreadLocal<Map<AMQShortString, WeakReference<AMQShortString>>> _localInternMap =
- new ThreadLocal<Map<AMQShortString, WeakReference<AMQShortString>>>()
- {
- protected Map<AMQShortString, WeakReference<AMQShortString>> initialValue()
- {
- return new WeakHashMap<AMQShortString, WeakReference<AMQShortString>>();
- };
- };
-
- private static final Map<AMQShortString, WeakReference<AMQShortString>> _globalInternMap =
- new WeakHashMap<AMQShortString, WeakReference<AMQShortString>>();
-
- private static final Logger _logger = LoggerFactory.getLogger(AMQShortString.class);
-
- private final byte[] _data;
- private final int _offset;
- private int _hashCode;
- private final int _length;
- private static final char[] EMPTY_CHAR_ARRAY = new char[0];
- private char[] chars;
- private String str;
-
- public static final AMQShortString EMPTY_STRING = new AMQShortString((String)null);
-
- public AMQShortString(byte[] data)
- {
-
- _data = data.clone();
- _length = data.length;
- _offset = 0;
- }
-
- public AMQShortString(byte[] data, int pos)
- {
- final int size = data[pos++];
- final byte[] dataCopy = new byte[size];
- System.arraycopy(data,pos,dataCopy,0,size);
- _length = size;
- _data = dataCopy;
- _offset = 0;
- }
-
- public AMQShortString(String data)
- {
- this((data == null) ? EMPTY_CHAR_ARRAY : data.toCharArray());
-
- }
-
- public AMQShortString(char[] data)
- {
- if (data == null)
- {
- throw new NullPointerException("Cannot create AMQShortString with null char[]");
- }
-
- final int length = data.length;
- final byte[] stringBytes = new byte[length];
- int hash = 0;
- for (int i = 0; i < length; i++)
- {
- stringBytes[i] = (byte) (0xFF & data[i]);
- hash = (31 * hash) + stringBytes[i];
- }
- _hashCode = hash;
- _data = stringBytes;
-
- _length = length;
- _offset = 0;
-
- }
-
- public AMQShortString(CharSequence charSequence)
- {
- final int length = charSequence.length();
- final byte[] stringBytes = new byte[length];
- int hash = 0;
- for (int i = 0; i < length; i++)
- {
- stringBytes[i] = ((byte) (0xFF & charSequence.charAt(i)));
- hash = (31 * hash) + stringBytes[i];
-
- }
-
- _data = stringBytes;
- _hashCode = hash;
- _length = length;
- _offset = 0;
-
- }
-
- private AMQShortString(ByteBuffer data, final int length)
- {
- byte[] dataBytes = new byte[length];
- data.get(dataBytes);
- _data = dataBytes;
- _length = length;
- _offset = 0;
-
- }
-
- private AMQShortString(final byte[] data, final int from, final int to)
- {
- _offset = from;
- _length = to - from;
- _data = data;
- }
-
-
- /**
- * Get the length of the short string
- * @return length of the underlying byte array
- */
- public int length()
- {
- return _length;
- }
-
- public char charAt(int index)
- {
-
- return (char) _data[_offset + index];
-
- }
-
- public CharSequence subSequence(int start, int end)
- {
- return new CharSubSequence(start, end);
- }
-
- public int writeToByteArray(byte[] encoding, int pos)
- {
- final int size = length();
- encoding[pos++] = (byte) size;
- System.arraycopy(_data,_offset,encoding,pos,size);
- return pos+size;
- }
-
- public static AMQShortString readFromByteArray(byte[] byteEncodedDestination, int pos)
- {
-
-
- final AMQShortString shortString = new AMQShortString(byteEncodedDestination, pos);
- if(shortString.length() == 0)
- {
- return null;
- }
- else
- {
- return shortString;
- }
- }
-
- public static AMQShortString readFromBuffer(ByteBuffer buffer)
- {
- final short length = buffer.getUnsigned();
- if (length == 0)
- {
- return null;
- }
- else
- {
-
- return new AMQShortString(buffer, length);
- }
- }
-
- public byte[] getBytes()
- {
- if(_offset == 0 && _length == _data.length)
- {
- return _data.clone();
- }
- else
- {
- byte[] data = new byte[_length];
- System.arraycopy(_data,_offset,data,0,_length);
- return data;
- }
- }
-
- public void writeToBuffer(ByteBuffer buffer)
- {
-
- final int size = length();
- //buffer.setAutoExpand(true);
- buffer.put((byte) size);
- buffer.put(_data, _offset, size);
- }
-
- private final class CharSubSequence implements CharSequence
- {
- private final int _sequenceOffset;
- private final int _end;
-
- public CharSubSequence(final int offset, final int end)
- {
- _sequenceOffset = offset;
- _end = end;
- }
-
- public int length()
- {
- return _end - _sequenceOffset;
- }
-
- public char charAt(int index)
- {
- return AMQShortString.this.charAt(index + _sequenceOffset);
- }
-
- public CharSequence subSequence(int start, int end)
- {
- return new CharSubSequence(start + _sequenceOffset, end + _sequenceOffset);
- }
- }
-
- public char[] asChars()
- {
- if (chars == null)
- {
- final int size = length();
- chars = new char[size];
-
- for (int i = 0; i < size; i++)
- {
- chars[i] = (char) _data[i + _offset];
- }
- }
- return chars;
- }
-
- public String asString()
- {
- if (str == null)
- {
- str = new String(asChars());
- }
-
- return str;
- }
-
- public boolean equals(Object o)
- {
-
-
- if(o instanceof AMQShortString)
- {
- return equals((AMQShortString)o);
- }
- if(o instanceof CharSequence)
- {
- return equals((CharSequence)o);
- }
-
- if (o == null)
- {
- return false;
- }
-
- if (o == this)
- {
- return true;
- }
-
-
- return false;
-
- }
-
- public boolean equals(final AMQShortString otherString)
- {
- if (otherString == this)
- {
- return true;
- }
-
- if (otherString == null)
- {
- return false;
- }
-
- if ((_hashCode != 0) && (otherString._hashCode != 0) && (_hashCode != otherString._hashCode))
- {
- return false;
- }
-
- return (_offset == 0 && otherString._offset == 0 && _length == _data.length && otherString._length == otherString._data.length && Arrays.equals(_data,otherString._data))
- || Arrays.equals(getBytes(),otherString.getBytes());
-
- }
-
- public boolean equals(CharSequence s)
- {
- if(s instanceof AMQShortString)
- {
- return equals((AMQShortString)s);
- }
-
- if (s == null)
- {
- return false;
- }
-
- if (s.length() != length())
- {
- return false;
- }
-
- for (int i = 0; i < length(); i++)
- {
- if (charAt(i) != s.charAt(i))
- {
- return false;
- }
- }
-
- return true;
- }
-
- public int hashCode()
- {
- int hash = _hashCode;
- if (hash == 0)
- {
- final int size = length();
-
- for (int i = 0; i < size; i++)
- {
- hash = (31 * hash) + _data[i+_offset];
- }
-
- _hashCode = hash;
- }
-
- return hash;
- }
-
- public void setDirty()
- {
- _hashCode = 0;
- }
-
- public String toString()
- {
- return asString();
- }
-
- public int compareTo(AMQShortString name)
- {
- if (name == null)
- {
- return 1;
- }
- else
- {
-
- if (name.length() < length())
- {
- return -name.compareTo(this);
- }
-
- for (int i = 0; i < length(); i++)
- {
- final byte d = _data[i+_offset];
- final byte n = name._data[i+name._offset];
- if (d < n)
- {
- return -1;
- }
-
- if (d > n)
- {
- return 1;
- }
- }
-
- return (length() == name.length()) ? 0 : -1;
- }
- }
-
- public AMQShortStringTokenizer tokenize(byte delim)
- {
- return new TokenizerImpl(delim);
- }
-
-
- public AMQShortString intern()
- {
-
- hashCode();
-
- Map<AMQShortString, WeakReference<AMQShortString>> localMap =
- _localInternMap.get();
-
- WeakReference<AMQShortString> ref = localMap.get(this);
- AMQShortString internString;
-
- if(ref != null)
- {
- internString = ref.get();
- if(internString != null)
- {
- return internString;
- }
- }
-
-
- synchronized(_globalInternMap)
- {
-
- ref = _globalInternMap.get(this);
- if((ref == null) || ((internString = ref.get()) == null))
- {
- internString = new AMQShortString(getBytes());
- ref = new WeakReference(internString);
- _globalInternMap.put(internString, ref);
- }
-
- }
- localMap.put(internString, ref);
- return internString;
-
- }
-
- private int occurences(final byte delim)
- {
- int count = 0;
- final int end = _offset + _length;
- for(int i = _offset ; i < end ; i++ )
- {
- if(_data[i] == delim)
- {
- count++;
- }
- }
- return count;
- }
-
- private int indexOf(final byte val, final int pos)
- {
-
- for(int i = pos; i < length(); i++)
- {
- if(_data[_offset+i] == val)
- {
- return i;
- }
- }
- return -1;
- }
-
-
- public static AMQShortString join(final Collection<AMQShortString> terms,
- final AMQShortString delim)
- {
- if(terms.size() == 0)
- {
- return EMPTY_STRING;
- }
-
- int size = delim.length() * (terms.size() - 1);
- for(AMQShortString term : terms)
- {
- size += term.length();
- }
-
- byte[] data = new byte[size];
- int pos = 0;
- final byte[] delimData = delim._data;
- final int delimOffset = delim._offset;
- final int delimLength = delim._length;
-
-
- for(AMQShortString term : terms)
- {
-
- if(pos!=0)
- {
- System.arraycopy(delimData, delimOffset,data,pos, delimLength);
- pos+=delimLength;
- }
- System.arraycopy(term._data,term._offset,data,pos,term._length);
- pos+=term._length;
- }
-
-
-
- return new AMQShortString(data,0,size);
- }
-
- public int toIntValue()
- {
- int pos = 0;
- int val = 0;
-
-
- boolean isNegative = (_data[pos] == MINUS);
- if(isNegative)
- {
- pos++;
- }
- while(pos < _length)
- {
- int digit = (int) (_data[pos++] - ZERO);
- if((digit < 0) || (digit > 9))
- {
- throw new NumberFormatException("\""+toString()+"\" is not a valid number");
- }
- val = val * 10;
- val += digit;
- }
- if(isNegative)
- {
- val = val * -1;
- }
- return val;
- }
-
- public boolean contains(final byte b)
- {
- for(int i = 0; i < _length; i++)
- {
- if(_data[i] == b)
- {
- return true;
- }
- }
- return false; //To change body of created methods use File | Settings | File Templates.
- }
-
-}
+/* + * + * 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.framing; + +import org.apache.mina.common.ByteBuffer; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.lang.ref.WeakReference; + +/** + * A short string is a representation of an AMQ Short String + * Short strings differ from the Java String class by being limited to on ASCII characters (0-127) + * and thus can be held more effectively in a byte buffer. + * + */ +public final class AMQShortString implements CharSequence, Comparable<AMQShortString> +{ + private static final byte MINUS = (byte)'-'; + private static final byte ZERO = (byte) '0'; + + + + private final class TokenizerImpl implements AMQShortStringTokenizer + { + private final byte _delim; + private int _count = -1; + private int _pos = 0; + + public TokenizerImpl(final byte delim) + { + _delim = delim; + } + + public int countTokens() + { + if(_count == -1) + { + _count = 1 + AMQShortString.this.occurences(_delim); + } + return _count; + } + + public AMQShortString nextToken() + { + if(_pos <= AMQShortString.this.length()) + { + int nextDelim = AMQShortString.this.indexOf(_delim, _pos); + if(nextDelim == -1) + { + nextDelim = AMQShortString.this.length(); + } + + AMQShortString nextToken = AMQShortString.this.substring(_pos, nextDelim++); + _pos = nextDelim; + return nextToken; + } + else + { + return null; + } + } + + public boolean hasMoreTokens() + { + return _pos <= AMQShortString.this.length(); + } + } + + private AMQShortString substring(final int from, final int to) + { + return new AMQShortString(_data, from, to); + } + + + private static final ThreadLocal<Map<AMQShortString, WeakReference<AMQShortString>>> _localInternMap = + new ThreadLocal<Map<AMQShortString, WeakReference<AMQShortString>>>() + { + protected Map<AMQShortString, WeakReference<AMQShortString>> initialValue() + { + return new WeakHashMap<AMQShortString, WeakReference<AMQShortString>>(); + }; + }; + + private static final Map<AMQShortString, WeakReference<AMQShortString>> _globalInternMap = + new WeakHashMap<AMQShortString, WeakReference<AMQShortString>>(); + + private static final Logger _logger = LoggerFactory.getLogger(AMQShortString.class); + + private final byte[] _data; + private final int _offset; + private int _hashCode; + private final int _length; + private static final char[] EMPTY_CHAR_ARRAY = new char[0]; + + public static final AMQShortString EMPTY_STRING = new AMQShortString((String)null); + + public AMQShortString(byte[] data) + { + + _data = data.clone(); + _length = data.length; + _offset = 0; + } + + public AMQShortString(byte[] data, int pos) + { + final int size = data[pos++]; + final byte[] dataCopy = new byte[size]; + System.arraycopy(data,pos,dataCopy,0,size); + _length = size; + _data = dataCopy; + _offset = 0; + } + + public AMQShortString(String data) + { + this((data == null) ? EMPTY_CHAR_ARRAY : data.toCharArray()); + + } + + public AMQShortString(char[] data) + { + if (data == null) + { + throw new NullPointerException("Cannot create AMQShortString with null char[]"); + } + + final int length = data.length; + final byte[] stringBytes = new byte[length]; + int hash = 0; + for (int i = 0; i < length; i++) + { + stringBytes[i] = (byte) (0xFF & data[i]); + hash = (31 * hash) + stringBytes[i]; + } + _hashCode = hash; + _data = stringBytes; + + _length = length; + _offset = 0; + + } + + public AMQShortString(CharSequence charSequence) + { + final int length = charSequence.length(); + final byte[] stringBytes = new byte[length]; + int hash = 0; + for (int i = 0; i < length; i++) + { + stringBytes[i] = ((byte) (0xFF & charSequence.charAt(i))); + hash = (31 * hash) + stringBytes[i]; + + } + + _data = stringBytes; + _hashCode = hash; + _length = length; + _offset = 0; + + } + + private AMQShortString(ByteBuffer data, final int length) + { + byte[] dataBytes = new byte[length]; + data.get(dataBytes); + _data = dataBytes; + _length = length; + _offset = 0; + + } + + private AMQShortString(final byte[] data, final int from, final int to) + { + _offset = from; + _length = to - from; + _data = data; + } + + + /** + * Get the length of the short string + * @return length of the underlying byte array + */ + public int length() + { + return _length; + } + + public char charAt(int index) + { + + return (char) _data[_offset + index]; + + } + + public CharSequence subSequence(int start, int end) + { + return new CharSubSequence(start, end); + } + + public int writeToByteArray(byte[] encoding, int pos) + { + final int size = length(); + encoding[pos++] = (byte) size; + System.arraycopy(_data,_offset,encoding,pos,size); + return pos+size; + } + + public static AMQShortString readFromByteArray(byte[] byteEncodedDestination, int pos) + { + + + final AMQShortString shortString = new AMQShortString(byteEncodedDestination, pos); + if(shortString.length() == 0) + { + return null; + } + else + { + return shortString; + } + } + + public static AMQShortString readFromBuffer(ByteBuffer buffer) + { + final short length = buffer.getUnsigned(); + if (length == 0) + { + return null; + } + else + { + + return new AMQShortString(buffer, length); + } + } + + public byte[] getBytes() + { + if(_offset == 0 && _length == _data.length) + { + return _data.clone(); + } + else + { + byte[] data = new byte[_length]; + System.arraycopy(_data,_offset,data,0,_length); + return data; + } + } + + public void writeToBuffer(ByteBuffer buffer) + { + + final int size = length(); + //buffer.setAutoExpand(true); + buffer.put((byte) size); + buffer.put(_data, _offset, size); + + } + + public boolean endsWith(String s) + { + return endsWith(new AMQShortString(s)); + } + + + public boolean endsWith(AMQShortString otherString) + { + + if (otherString.length() > length()) + { + return false; + } + + + int thisLength = length(); + int otherLength = otherString.length(); + + for (int i = 1; i <= otherLength; i++) + { + if (charAt(thisLength - i) != otherString.charAt(otherLength - i)) + { + return false; + } + } + return true; + } + + public boolean startsWith(String s) + { + return startsWith(new AMQShortString(s)); + } + + public boolean startsWith(AMQShortString otherString) + { + + if (otherString.length() > length()) + { + return false; + } + + for (int i = 0; i < otherString.length(); i++) + { + if (_data[i] != otherString._data[i]) + { + return false; + } + } + + return true; + + } + + public boolean startsWith(CharSequence otherString) + { + if (otherString.length() > length()) + { + return false; + } + + for (int i = 0; i < otherString.length(); i++) + { + if (charAt(i) != otherString.charAt(i)) + { + return false; + } + } + + return true; + } + + + private final class CharSubSequence implements CharSequence + { + private final int _sequenceOffset; + private final int _end; + + public CharSubSequence(final int offset, final int end) + { + _sequenceOffset = offset; + _end = end; + } + + public int length() + { + return _end - _sequenceOffset; + } + + public char charAt(int index) + { + return AMQShortString.this.charAt(index + _sequenceOffset); + } + + public CharSequence subSequence(int start, int end) + { + return new CharSubSequence(start + _sequenceOffset, end + _sequenceOffset); + } + } + + public char[] asChars() + { + final int size = length(); + final char[] chars = new char[size]; + + for (int i = 0; i < size; i++) + { + chars[i] = (char) _data[i + _offset]; + } + + return chars; + } + + public String asString() + { + return new String(asChars()); + } + + public boolean equals(Object o) + { + + + if(o instanceof AMQShortString) + { + return equals((AMQShortString)o); + } + if(o instanceof CharSequence) + { + return equals((CharSequence)o); + } + + if (o == null) + { + return false; + } + + if (o == this) + { + return true; + } + + + return false; + + } + + public boolean equals(final AMQShortString otherString) + { + if (otherString == this) + { + return true; + } + + if (otherString == null) + { + return false; + } + + if ((_hashCode != 0) && (otherString._hashCode != 0) && (_hashCode != otherString._hashCode)) + { + return false; + } + + return (_offset == 0 && otherString._offset == 0 && _length == _data.length && otherString._length == otherString._data.length && Arrays.equals(_data,otherString._data)) + || Arrays.equals(getBytes(),otherString.getBytes()); + + } + + public boolean equals(CharSequence s) + { + if(s instanceof AMQShortString) + { + return equals((AMQShortString)s); + } + + if (s == null) + { + return false; + } + + if (s.length() != length()) + { + return false; + } + + for (int i = 0; i < length(); i++) + { + if (charAt(i) != s.charAt(i)) + { + return false; + } + } + + return true; + } + + public int hashCode() + { + int hash = _hashCode; + if (hash == 0) + { + final int size = length(); + + for (int i = 0; i < size; i++) + { + hash = (31 * hash) + _data[i+_offset]; + } + + _hashCode = hash; + } + + return hash; + } + + public void setDirty() + { + _hashCode = 0; + } + + public String toString() + { + return asString(); + } + + public int compareTo(AMQShortString name) + { + if (name == null) + { + return 1; + } + else + { + + if (name.length() < length()) + { + return -name.compareTo(this); + } + + for (int i = 0; i < length(); i++) + { + final byte d = _data[i+_offset]; + final byte n = name._data[i+name._offset]; + if (d < n) + { + return -1; + } + + if (d > n) + { + return 1; + } + } + + return (length() == name.length()) ? 0 : -1; + } + } + + + public AMQShortStringTokenizer tokenize(byte delim) + { + return new TokenizerImpl(delim); + } + + + public AMQShortString intern() + { + + hashCode(); + + Map<AMQShortString, WeakReference<AMQShortString>> localMap = + _localInternMap.get(); + + WeakReference<AMQShortString> ref = localMap.get(this); + AMQShortString internString; + + if(ref != null) + { + internString = ref.get(); + if(internString != null) + { + return internString; + } + } + + + synchronized(_globalInternMap) + { + + ref = _globalInternMap.get(this); + if((ref == null) || ((internString = ref.get()) == null)) + { + internString = new AMQShortString(getBytes()); + ref = new WeakReference(internString); + _globalInternMap.put(internString, ref); + } + + } + localMap.put(internString, ref); + return internString; + + } + + private int occurences(final byte delim) + { + int count = 0; + final int end = _offset + _length; + for(int i = _offset ; i < end ; i++ ) + { + if(_data[i] == delim) + { + count++; + } + } + return count; + } + + private int indexOf(final byte val, final int pos) + { + + for(int i = pos; i < length(); i++) + { + if(_data[_offset+i] == val) + { + return i; + } + } + return -1; + } + + + public static AMQShortString join(final Collection<AMQShortString> terms, + final AMQShortString delim) + { + if(terms.size() == 0) + { + return EMPTY_STRING; + } + + int size = delim.length() * (terms.size() - 1); + for(AMQShortString term : terms) + { + size += term.length(); + } + + byte[] data = new byte[size]; + int pos = 0; + final byte[] delimData = delim._data; + final int delimOffset = delim._offset; + final int delimLength = delim._length; + + + for(AMQShortString term : terms) + { + + if(pos!=0) + { + System.arraycopy(delimData, delimOffset,data,pos, delimLength); + pos+=delimLength; + } + System.arraycopy(term._data,term._offset,data,pos,term._length); + pos+=term._length; + } + + + + return new AMQShortString(data,0,size); + } + + public int toIntValue() + { + int pos = 0; + int val = 0; + + + boolean isNegative = (_data[pos] == MINUS); + if(isNegative) + { + pos++; + } + while(pos < _length) + { + int digit = (int) (_data[pos++] - ZERO); + if((digit < 0) || (digit > 9)) + { + throw new NumberFormatException("\""+toString()+"\" is not a valid number"); + } + val = val * 10; + val += digit; + } + if(isNegative) + { + val = val * -1; + } + return val; + } + + public boolean contains(final byte b) + { + for(int i = 0; i < _length; i++) + { + if(_data[i] == b) + { + return true; + } + } + return false; //To change body of created methods use File | Settings | File Templates. + } + +} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/framing/AMQShortStringTest.java b/qpid/java/common/src/test/java/org/apache/qpid/framing/AMQShortStringTest.java new file mode 100644 index 0000000000..0ea2c8b9c1 --- /dev/null +++ b/qpid/java/common/src/test/java/org/apache/qpid/framing/AMQShortStringTest.java @@ -0,0 +1,62 @@ +/* + * 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.framing; + +import junit.framework.TestCase; +public class AMQShortStringTest extends TestCase +{ + + AMQShortString Hello = new AMQShortString("Hello"); + AMQShortString Hell = new AMQShortString("Hell"); + AMQShortString Goodbye = new AMQShortString("Goodbye"); + AMQShortString Good = new AMQShortString("Good"); + AMQShortString Bye = new AMQShortString("Bye"); + + public void testStartsWith() + { + assertTrue(Hello.startsWith(Hell)); + + assertFalse(Hell.startsWith(Hello)); + + assertTrue(Goodbye.startsWith(Good)); + + assertFalse(Good.startsWith(Goodbye)); + } + + public void testEndWith() + { + assertFalse(Hell.endsWith(Hello)); + + assertTrue(Goodbye.endsWith(new AMQShortString("bye"))); + + assertFalse(Goodbye.endsWith(Bye)); + } + + + public void testEquals() + { + assertEquals(Goodbye, new AMQShortString("Goodbye")); + assertEquals(new AMQShortString("A"), new AMQShortString("A")); + assertFalse(new AMQShortString("A").equals(new AMQShortString("a"))); + } + + +} diff --git a/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/IsolatedClassLoader.java b/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/IsolatedClassLoader.java index f9c1ff5e9b..dc74590f60 100644 --- a/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/IsolatedClassLoader.java +++ b/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/IsolatedClassLoader.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.maven;
diff --git a/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/TKTestScriptGenMojo.java b/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/TKTestScriptGenMojo.java index be665da018..5c7669e069 100644 --- a/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/TKTestScriptGenMojo.java +++ b/qpid/java/junit-toolkit-maven-plugin/src/main/org/apache/qpid/junit/maven/TKTestScriptGenMojo.java @@ -1,20 +1,28 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.maven;
+import org.apache.maven.plugin.AbstractMojo;
+import org.apache.maven.plugin.MojoExecutionException;
+
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
@@ -22,8 +30,6 @@ import java.io.Writer; import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.maven.plugin.AbstractMojo;
-import org.apache.maven.plugin.MojoExecutionException;
/**
* <p><table id="crc"><caption>CRC Card</caption>
@@ -111,8 +117,7 @@ public class TKTestScriptGenMojo extends AbstractMojo }
commandLine +=
- "${JAVA_OPTS} -cp " + testJar + " org.apache.qpid.junit.extensions.TKTestRunner " + testOptions
- + " ${ARGS}";
+ "${JAVA_OPTS} -cp " + testJar + " org.apache.qpid.junit.extensions.TKTestRunner " + testOptions + " ${ARGS}";
getLog().info("Generating Script for test: " + testName);
getLog().debug(commandLine);
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/BaseThrottle.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/BaseThrottle.java index 17d1fb7692..e8e203f0a3 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/BaseThrottle.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/BaseThrottle.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions;
@@ -58,7 +63,7 @@ public abstract class BaseThrottle implements Throttle }
// Calculate the cycle time.
- cycleTimeNanos = (long)(1000000000f / hertz);
+ cycleTimeNanos = (long) (1000000000f / hertz);
// Reset the first pass flag.
firstCall = false;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/DurationTestDecorator.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/DurationTestDecorator.java index 187fb21e9a..fe1e044e67 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/DurationTestDecorator.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/DurationTestDecorator.java @@ -1,28 +1,33 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions;
-import java.util.Timer;
-import java.util.TimerTask;
-
import junit.framework.Test;
import junit.framework.TestResult;
import org.apache.log4j.Logger;
+import java.util.Timer;
+import java.util.TimerTask;
+
/**
* A test decorator that runs a test repeatedly until a specified length of time has passed.
*
@@ -95,7 +100,7 @@ public class DurationTestDecorator extends WrappedSuiteTestDecorator implements if (testResult instanceof TKTestResult)
{
- tkTestResult = (TKTestResult)testResult;
+ tkTestResult = (TKTestResult) testResult;
}
// Work out when the test should end.
@@ -111,7 +116,7 @@ public class DurationTestDecorator extends WrappedSuiteTestDecorator implements log.debug("Creating duration timer.");
durationTimer = new Timer();
- durationTimer.schedule(new DurationTimerTask((TKTestResult)testResult), duration);
+ durationTimer.schedule(new DurationTimerTask((TKTestResult) testResult), duration);
}
// Run the test until the duration times out or the shutdown flag is set. The test method may not exit until
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/SetupTaskAware.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/SetupTaskAware.java index 320a2c2ac5..0e8e1879b6 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/SetupTaskAware.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/SetupTaskAware.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/CommandLineParser.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/CommandLineParser.java index 744d59d905..61c58bf3ba 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/CommandLineParser.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/CommandLineParser.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
@@ -589,7 +594,7 @@ public class CommandLineParser {
for (Object propKey : trailingProperties.keySet())
{
- String name = (String)propKey;
+ String name = (String) propKey;
String value = trailingProperties.getProperty(name);
properties.setProperty(name, value);
@@ -609,7 +614,7 @@ public class CommandLineParser {
for (Object propKey : parsedProperties.keySet())
{
- String name = (String)propKey;
+ String name = (String) propKey;
String value = parsedProperties.getProperty(name);
// This filters out all trailing items.
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ContextualProperties.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ContextualProperties.java index f4c58654d1..cabbf7869a 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ContextualProperties.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ContextualProperties.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/MathUtils.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/MathUtils.java index 3db9eb6793..7a45632643 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/MathUtils.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/MathUtils.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
@@ -413,7 +418,7 @@ public class MathUtils */
private static void roundAndAdd(ArrayList<Integer> result, double value)
{
- int roundedValue = (int)Math.round(value);
+ int roundedValue = (int) Math.round(value);
if (!result.contains(roundedValue))
{
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ParsedProperties.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ParsedProperties.java index ffb5df0751..59c8cfbd3a 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ParsedProperties.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/ParsedProperties.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/SizeOf.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/SizeOf.java index 715cafde87..5f3ebb4545 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/SizeOf.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/SizeOf.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/StackQueue.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/StackQueue.java index 45a7854646..9078c0e247 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/StackQueue.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/StackQueue.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestContextProperties.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestContextProperties.java index fb287d03f8..edb7b6d73a 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestContextProperties.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestContextProperties.java @@ -1,17 +1,22 @@ /*
- * Copyright 2007 Rupert Smith.
*
- * Licensed 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
+ * 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
+ * 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.
*
- * 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.junit.extensions.util;
@@ -92,7 +97,7 @@ public class TestContextProperties extends ParsedProperties for (Object key : defaults.keySet())
{
- String stringKey = (String)key;
+ String stringKey = (String) key;
String value = defaults.getProperty(stringKey);
props.setPropertyIfNull(stringKey, value);
diff --git a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestUtils.java b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestUtils.java index fcc2f0ab94..553a41ecae 100644 --- a/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestUtils.java +++ b/qpid/java/junit-toolkit/src/main/org/apache/qpid/junit/extensions/util/TestUtils.java @@ -1,17 +1,22 @@ /* - * Copyright 2007 Rupert Smith. * - * Licensed 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 + * 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 + * 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. * - * 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.junit.extensions.util; diff --git a/qpid/java/systests/etc/acl.config.xml b/qpid/java/systests/etc/acl.config.xml new file mode 100644 index 0000000000..3eb3a0335b --- /dev/null +++ b/qpid/java/systests/etc/acl.config.xml @@ -0,0 +1,229 @@ +<?xml version="1.0" encoding="ISO-8859-1"?> +<!-- + - + - 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. + - + --> +<broker> + <prefix>${QPID_HOME}</prefix> + <work>${QPID_WORK}</work> + <conf>${prefix}/etc</conf> + <connector> + <!-- Uncomment out this block and edit the keystorePath and keystorePassword + to enable SSL support + <ssl> + <enabled>true</enabled> + <sslOnly>true</sslOnly> + <keystorePath>/path/to/keystore.ks</keystorePath> + <keystorePassword>keystorepass</keystorePassword> + </ssl>--> + <qpidnio>false</qpidnio> + <!-- I've had the 0.0 and 0.1 Reader threads continually throwing IOException when client closes--> + <protectio>false</protectio> + <transport>nio</transport> + <port>5672</port> + <sslport>8672</sslport> + <socketReceiveBuffer>32768</socketReceiveBuffer> + <socketSendBuffer>32768</socketSendBuffer> + </connector> + <management> + <enabled>false</enabled> + <jmxport>8999</jmxport> + <security-enabled>false</security-enabled> + </management> + <advanced> + <filterchain enableExecutorPool="true"/> + <enablePooledAllocator>false</enablePooledAllocator> + <enableDirectBuffers>false</enableDirectBuffers> + <framesize>65535</framesize> + <compressBufferOnQueue>false</compressBufferOnQueue> + <enableJMSXUserID>true</enableJMSXUserID> + </advanced> + + <security> + <principal-databases> + <!-- Example use of Base64 encoded MD5 hashes for authentication via CRAM-MD5-Hashed --> + <principal-database> + <name>passwordfile</name> + <class>org.apache.qpid.server.security.auth.database.PlainPasswordFilePrincipalDatabase</class> + <attributes> + <attribute> + <name>passwordFile</name> + <value>${conf}/passwd</value> + </attribute> + </attributes> + </principal-database> + </principal-databases> + + <access> + <class>org.apache.qpid.server.security.access.plugins.DenyAll</class> + </access> + + <jmx> + <access>${conf}/jmxremote.access</access> + <principal-database>passwordfile</principal-database> + </jmx> + </security> + + <virtualhosts> + <directory>${conf}/virtualhosts</directory> + + <virtualhost> + <name>test</name> + <test> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + + <queues> + <exchange>amq.direct</exchange> + <!-- 4Mb --> + <maximumQueueDepth>4235g264</maximumQueueDepth> + <!-- 2Mb --> + <maximumMessageSize>2117632</maximumMessageSize> + <!-- 10 mins --> + <maximumMessageAge>600000</maximumMessageAge> + </queues> + + + <security> + <access> + <class>org.apache.qpid.server.security.access.plugins.SimpleXML</class> + </access> + + <access_control_list> + <!-- This section grants pubish rights to an exchange + routing key pair --> + <publish> + <exchanges> + <exchange> + <name>amq.direct</name> + <routing_keys> + + <!-- Allow clients to publish requests --> + <routing_key> + <value>example.RequestQueue</value> + <users> + <user>client</user> + </users> + </routing_key> + + <!-- Allow the processor to respond to a client on their Temporary Topic --> + <routing_key> + <value>tmp_*</value> + <users> + <user>server</user> + </users> + </routing_key> + </routing_keys> + + </exchange> + </exchanges> + </publish> + + <!-- This section grants users the ability to consume from the broker --> + <consume> + <queues> + + <!-- Allow the clients to consume from their temporary queues--> + <queue> + <temporary/> + <users> + <user>client</user> + </users> + </queue> + + + <!-- Only allow the server to consume from the Request Queue--> + <queue> + <name>example.RequestQueue</name> + <users> + <user>server</user> + </users> + </queue> + + + </queues> + </consume> + + <!-- This section grants clients the ability to create queues and exchanges --> + <create> + <queues> + <!-- Allow clients to create temporary queues--> + <queue> + <temporary/> + <exchanges> + <exchange> + <name>amq.direct</name> + <users> + <user>client</user> + </users> + </exchange> + </exchanges> + </queue> + <!-- Allow the server to create the Request Queue--> + <queue> + <name>example.RequestQueue</name> + <users> + <user>server</user> + </users> + </queue> + + </queues> + </create> + + + </access_control_list> + + </security> + </test> + </virtualhost> + + + <virtualhost> + <name>development</name> + <development> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + </development> + </virtualhost> + + <virtualhost> + <name>localhost</name> + <localhost> + <store> + <class>org.apache.qpid.server.store.MemoryMessageStore</class> + </store> + </localhost> + </virtualhost> + + </virtualhosts> + + <heartbeat> + <delay>0</delay> + <timeoutFactor>2.0</timeoutFactor> + </heartbeat> + + <queue> + <auto_register>false</auto_register> + </queue> + + <virtualhosts>${conf}/virtualhosts.xml</virtualhosts> +</broker> + + diff --git a/qpid/java/systests/pom.xml b/qpid/java/systests/pom.xml index f42ab3f49d..62e7e50136 100644 --- a/qpid/java/systests/pom.xml +++ b/qpid/java/systests/pom.xml @@ -6,9 +6,9 @@ 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 @@ -62,11 +62,11 @@ </dependency> <!-- Test Dependencies --> - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - <version>1.4.0</version> - <scope>test</scope> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + <version>1.4.0</version> + <scope>test</scope> </dependency> </dependencies> @@ -82,18 +82,26 @@ <includes> <include>**/*Test.class</include> </includes> - + <systemProperties> <property> <name>example.plugin.target</name> <value>${basedir}/${topDirectoryLocation}/plugins/target</value> </property> + <property> + <name>QPID_EXAMPLE_HOME</name> + <value>${basedir}</value> + </property> + <property> + <name>QPID_HOME</name> + <value>${basedir}/${topDirectoryLocation}/broker</value> + </property> </systemProperties> - + <excludes> <exclude>**/testcases/ImmediateMessageTest.class</exclude> <exclude>**/testcases/MandatoryMessageTest.class</exclude> - <exclude>**/testcases/RollbackTest.class</exclude> + <exclude>**/testcases/RollbackTest.class</exclude> <exclude>**/testcases/TTLTest.class</exclude> <exclude>**/testcases/FailoverTest.class</exclude> </excludes> @@ -112,23 +120,23 @@ <value>${log4j.configuration}</value> </property> </systemproperties> - + <testrunner>org.apache.qpid.junit.extensions.TKTestRunner</testrunner> - + <testrunneroptions> <option>-X:decorators "org.apache.qpid.test.framework.qpid.InVMBrokerDecorator:org.apache.qpid.test.framework.qpid.AMQPFeatureDecorator"</option> <!--<option>-d30S</option>--> <option>-o ${basedir}/target/surefire-reports</option> <option>--xml</option> </testrunneroptions> - + <testrunnerproperties> <property> <name>notApplicableAssertion</name> <value>warn</value> </property> </testrunnerproperties> - + <commands> <AMQBrokerManagerMBeanTest>-n AMQBrokerManagerMBeanTest org.apache.qpid.server.AMQBrokerManagerMBeanTest </AMQBrokerManagerMBeanTest> <TxAckTest>-n TxAckTest org.apache.qpid.server.ack.TxAckTest </TxAckTest> @@ -153,16 +161,16 @@ <!--<Mandatory-Message-Test>-n Mandatory-Test -s[1] org.apache.qpid.test.testcases.MandatoryMessageTest</Mandatory-Message-Test>--> <!--<Rollback-Test>-n Rollback-Test -s[1] org.apache.qpid.test.testcases.RollbackTest</Rollback-Test>--> </commands> - + </configuration> - <executions> + <executions> <execution> <id>framework_tests</id> - <phase>test</phase> + <phase>test</phase> <goals> <goal>tktest</goal> - </goals> + </goals> </execution> </executions> </plugin> diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java new file mode 100644 index 0000000000..d2c677c8a5 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java @@ -0,0 +1,581 @@ +/* + * 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.acl; + +import junit.framework.TestCase; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.client.*; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; +import org.apache.qpid.AMQException; +import org.apache.qpid.jms.ConnectionListener; +import org.apache.qpid.url.URLSyntaxException; + +import javax.jms.*; +import java.io.File; + + +public class SimpleACLTest extends TestCase implements ConnectionListener +{ + private String BROKER = "vm://:1";//"tcp://localhost:5672"; + + public void setUp() throws Exception + { + // Initialise ACLs. + final String QpidExampleHome = System.getProperty("QPID_EXAMPLE_HOME"); + final File defaultaclConfigFile = new File(QpidExampleHome, "etc/acl.config.xml"); + + if (!defaultaclConfigFile.exists() || System.getProperty("QPID_HOME") == null) + { + System.err.println("Configuration file not found:" + defaultaclConfigFile); + fail("Configuration file not found:" + defaultaclConfigFile); + } + + ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(defaultaclConfigFile); + + ApplicationRegistry.initialise(config, 1); + + TransportConnection.createVMBroker(1); + } + + public void tearDown() + { + ApplicationRegistry.remove(1); + TransportConnection.killAllVMBrokers(); + } + + public String createConnectionString(String username, String password, String broker) + { + + return "amqp://" + username + ":" + password + "@clientid/test?brokerlist='" + broker + "'"; + } + + public void testAccessAuthorized() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + Session sesh = conn.createSession(true, Session.SESSION_TRANSACTED); + + conn.start(); + + //Do something to show connection is active. + sesh.rollback(); + + conn.close(); + } + catch (Exception e) + { + fail("Connection was not created due to:" + e.getMessage()); + } + } + + public void testAccessNoRights() throws URLSyntaxException, JMSException + { + try + { + Connection conn = new AMQConnection(createConnectionString("guest", "guest", BROKER)); + + //Attempt to do do things to test connection. + Session sesh = conn.createSession(true, Session.SESSION_TRANSACTED); + conn.start(); + sesh.rollback(); + + conn.close(); + fail("Connection was created."); + } + catch (AMQException amqe) + { + if (amqe.getCause() instanceof Exception) + { + System.err.println("QPID-594 : WARNING RACE CONDITION. Unable to determine cause of Connection Failure."); + } + assertEquals("Linked Exception Incorrect", JMSException.class, amqe.getCause().getClass()); + Exception linked = ((JMSException) amqe.getCause()).getLinkedException(); + assertEquals("Exception was wrong type", AMQAuthenticationException.class, linked.getClass()); + assertEquals("Incorrect error code thrown", 403, ((AMQAuthenticationException) linked).getErrorCode().getCode()); + } + } + + public void testClientConsumeFromTempQueueValid() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + sesh.createConsumer(sesh.createTemporaryQueue()); + + conn.close(); + } + catch (Exception e) + { + fail("Test failed due to:" + e.getMessage()); + } + } + + public void testClientConsumeFromNamedQueueInvalid() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + //Prevent Failover + ((AMQConnection) conn).setConnectionListener(this); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + sesh.createConsumer(sesh.createQueue("IllegalQueue")); + fail("Test failed as consumer was created."); + } + catch (JMSException e) + { + Throwable cause = e.getLinkedException(); + + assertNotNull("There was no liked exception", cause); + assertEquals("Wrong linked exception type", AMQAuthenticationException.class, cause.getClass()); + assertEquals("Incorrect error code received", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); + } + } + + public void testClientCreateTemporaryQueue() throws JMSException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + //Create Temporary Queue + ((AMQSession) sesh).declareQueue((AMQDestination) sesh.createTemporaryQueue(), + ((AMQSession) sesh).getProtocolHandler()); + + conn.close(); + } + catch (Exception e) + { + fail("Test failed due to:" + e.getMessage()); + } + } + + public void testClientCreateNamedQueue() throws JMSException, URLSyntaxException, AMQException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + //Create a Named Queue + ((AMQSession) sesh).declareQueue((AMQDestination) sesh.createQueue("IllegalQueue"), + ((AMQSession) sesh).getProtocolHandler()); + + fail("Test failed as Queue creation succeded."); + } + catch (AMQAuthenticationException amqe) + { + assertEquals("Incorrect error code thrown", 403, ((AMQAuthenticationException) amqe).getErrorCode().getCode()); + } + } + + public void testClientPublishUsingTransactionSuccess() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + ((AMQConnection) conn).setConnectionListener(this); + + Session sesh = conn.createSession(true, Session.SESSION_TRANSACTED); + + conn.start(); + + MessageProducer sender = sesh.createProducer(sesh.createQueue("example.RequestQueue")); + + sender.send(sesh.createTextMessage("test")); + + //Send the message using a transaction as this will allow us to retrieve any errors that occur on the broker. + sesh.commit(); + + conn.close(); + } + catch (Exception e) + { + fail("Test publish failed:" + e); + } + } + + public void testClientPublishValidQueueSuccess() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + ((AMQConnection) conn).setConnectionListener(this); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + MessageProducer sender = ((AMQSession) sesh).createProducer(null); + + Queue queue = sesh.createQueue("example.RequestQueue"); + + // Send a message that we will wait to be sent, this should give the broker time to process the msg + // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not + // queue existence. + ((org.apache.qpid.jms.MessageProducer) sender).send(queue, sesh.createTextMessage("test"), + DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true); + + conn.close(); + } + catch (Exception e) + { + fail("Test publish failed:" + e); + } + } + + public void testClientPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + ((AMQConnection) conn).setConnectionListener(this); + + Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + MessageProducer sender = ((AMQSession) session).createProducer(null); + + Queue queue = session.createQueue("Invalid"); + + // Send a message that we will wait to be sent, this should give the broker time to close the connection + // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not + // queue existence. + ((org.apache.qpid.jms.MessageProducer) sender).send(queue, session.createTextMessage("test"), + DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true); + + // Test the connection with a valid consumer + session.createConsumer(session.createTemporaryQueue()).close(); + + //Connection should now be closed and will throw the exception caused by the above send + conn.close(); + + fail("Close is not expected to succeed."); + } + catch (JMSException e) + { + Throwable cause = e.getLinkedException(); + assertEquals("Incorrect exception", AMQAuthenticationException.class, cause.getClass()); + assertEquals("Incorrect error code thrown", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); + } + } + + public void testServerConsumeFromNamedQueueValid() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + sesh.createConsumer(sesh.createQueue("example.RequestQueue")); + + conn.close(); + } + catch (Exception e) + { + fail("Test failed due to:" + e.getMessage()); + } + } + + public void testServerConsumeFromNamedQueueInvalid() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + sesh.createConsumer(sesh.createQueue("Invalid")); + + fail("Test failed as consumer was created."); + } + catch (JMSException e) + { + Throwable cause = e.getLinkedException(); + + assertNotNull("There was no liked exception", cause); + assertEquals("Wrong linked exception type", AMQAuthenticationException.class, cause.getClass()); + assertEquals("Incorrect error code received", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); + } + } + + public void testServerConsumeFromTemporaryQueue() throws AMQException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + //Prevent Failover + ((AMQConnection) conn).setConnectionListener(this); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + sesh.createConsumer(sesh.createTemporaryQueue()); + fail("Test failed as consumer was created."); + } + catch (JMSException e) + { + Throwable cause = e.getLinkedException(); + + assertNotNull("There was no liked exception", cause); + assertEquals("Wrong linked exception type", AMQAuthenticationException.class, cause.getClass()); + assertEquals("Incorrect error code received", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); + } + } + + public void testServerCreateNamedQueueValid() throws JMSException, URLSyntaxException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + //Create Temporary Queue + ((AMQSession) sesh).declareQueue((AMQDestination) sesh.createQueue("example.RequestQueue"), + ((AMQSession) sesh).getProtocolHandler()); + + conn.close(); + } + catch (Exception e) + { + fail("Test failed due to:" + e.getMessage()); + } + } + + public void testServerCreateNamedQueueInValid() throws JMSException, URLSyntaxException, AMQException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + //Create a Named Queue + ((AMQSession) sesh).declareQueue((AMQDestination) sesh.createQueue("IllegalQueue"), + ((AMQSession) sesh).getProtocolHandler()); + + fail("Test failed as creation succeded."); + } + catch (AMQAuthenticationException amqe) + { + assertEquals("Incorrect error code thrown", 403, amqe.getErrorCode().getCode()); + } + } + + public void testServerCreateTemporyQueueInvalid() throws JMSException, URLSyntaxException, AMQException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + ((AMQSession) sesh).declareQueue((AMQDestination) sesh.createTemporaryQueue(), + ((AMQSession) sesh).getProtocolHandler()); + + fail("Test failed as creation succeded."); + } + catch (AMQAuthenticationException amqe) + { + assertEquals("Incorrect error code thrown", 403, amqe.getErrorCode().getCode()); + } + } + + /** + * This test uses both the cilent and sender to validate that the Server is able to publish to a temporary queue. + * The reason the client must be in volved is that the Serve is unable to create its own Temporary Queues. + * + * @throws AMQException + * @throws URLSyntaxException + * @throws JMSException + */ + public void testServerPublishUsingTransactionSuccess() throws AMQException, URLSyntaxException, JMSException + { + //Set up the Server + Connection serverConnection = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + ((AMQConnection) serverConnection).setConnectionListener(this); + + Session serverSession = serverConnection.createSession(true, Session.SESSION_TRANSACTED); + + Queue requestQueue = serverSession.createQueue("example.RequestQueue"); + + MessageConsumer server = serverSession.createConsumer(requestQueue); + + serverConnection.start(); + + //Set up the consumer + Connection clientConnection = new AMQConnection(createConnectionString("client", "guest", BROKER)); + + //Send a test mesage + Session clientSession = clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + Queue responseQueue = clientSession.createTemporaryQueue(); + + MessageConsumer clientResponse = clientSession.createConsumer(responseQueue); + + clientConnection.start(); + + Message request = clientSession.createTextMessage("Request"); + + assertNotNull("Response Queue is null", responseQueue); + + request.setJMSReplyTo(responseQueue); + + clientSession.createProducer(requestQueue).send(request); + + try + { + Message msg = null; + + msg = server.receive(2000); + + while (msg != null && !((TextMessage) msg).getText().equals("Request")) + { + msg = server.receive(2000); + } + + assertNotNull("Message not received", msg); + + assertNotNull("Reply-To is Null", msg.getJMSReplyTo()); + + MessageProducer sender = serverSession.createProducer(msg.getJMSReplyTo()); + + sender.send(serverSession.createTextMessage("Response")); + + //Send the message using a transaction as this will allow us to retrieve any errors that occur on the broker. + serverSession.commit(); + + serverConnection.close(); + + //Ensure Response is received. + Message clientResponseMsg = clientResponse.receive(2000); + assertNotNull("Client did not receive response message,", clientResponseMsg); + assertEquals("Incorrect message received", "Response", ((TextMessage) clientResponseMsg).getText()); + + } + catch (Exception e) + { + fail("Test publish failed:" + e); + } + } + + public void testServerPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException + { + try + { + Connection conn = new AMQConnection(createConnectionString("server", "guest", BROKER)); + + ((AMQConnection) conn).setConnectionListener(this); + + Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + conn.start(); + + MessageProducer sender = ((AMQSession) session).createProducer(null); + + Queue queue = session.createQueue("Invalid"); + + // Send a message that we will wait to be sent, this should give the broker time to close the connection + // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not + // queue existence. + ((org.apache.qpid.jms.MessageProducer) sender).send(queue, session.createTextMessage("test"), + DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true); + + // Test the connection with a valid consumer + session.createConsumer(session.createQueue("example.RequestQueue")).close(); + + //Connection should now be closed and will throw the exception caused by the above send + conn.close(); + + fail("Close is not expected to succeed."); + } + catch (JMSException e) + { + Throwable cause = e.getLinkedException(); + assertEquals("Incorrect exception", AMQAuthenticationException.class, cause.getClass()); + assertEquals("Incorrect error code thrown", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); + } + } + + // Connection Listener Interface - Used here to block failover + + public void bytesSent(long count) + { + } + + public void bytesReceived(long count) + { + } + + public boolean preFailover(boolean redirect) + { + //Prevent failover. + return false; + } + + public boolean preResubscribe() + { + return false; + } + + public void failoverComplete() + { + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java index 0218109369..83b4665be6 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java @@ -31,8 +31,8 @@ import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager; import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager; import org.apache.qpid.server.security.auth.database.PropertiesPrincipalDatabaseManager; -import org.apache.qpid.server.security.access.AccessManager; -import org.apache.qpid.server.security.access.AllowAll; +import org.apache.qpid.server.security.access.ACLPlugin; +import org.apache.qpid.server.security.access.plugins.AllowAll; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.TestableMemoryMessageStore; import org.apache.qpid.server.virtualhost.VirtualHost; @@ -54,7 +54,7 @@ public class TestApplicationRegistry extends ApplicationRegistry private ManagedObjectRegistry _managedObjectRegistry; - private AccessManager _accessManager; + private ACLPlugin _accessManager; private PrincipalDatabaseManager _databaseManager; @@ -137,11 +137,16 @@ public class TestApplicationRegistry extends ApplicationRegistry return null; //To change body of implemented methods use File | Settings | File Templates. } - public AccessManager getAccessManager() + public ACLPlugin getAccessManager() { return _accessManager; } + public void setAccessManager(ACLPlugin newManager) + { + _accessManager = newManager; + } + public MessageStore getMessageStore() { return _messageStore; |