This is an automated email from the ASF dual-hosted git repository.
orudyy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
The following commit(s) were added to refs/heads/master by this push:
new 1dd69d2 QPID-8369: [Broker-J] Add ability to limit a number of
connections per user
1dd69d2 is described below
commit 1dd69d26885abc6cf25b7c787d8b3bc727b2c74a
Author: Alex Rudyy <[email protected]>
AuthorDate: Fri Feb 28 14:08:44 2020 +0000
QPID-8369: [Broker-J] Add ability to limit a number of connections per user
This closes #43
---
.../apache/qpid/server/security/QpidPrincipal.java | 2 +-
.../security/auth/manager/SpnegoAuthenticator.java | 44 ++++-
.../qpid/server/transport/AMQPConnection.java | 7 +
.../server/transport/AbstractAMQPConnection.java | 28 ++++
.../server/virtualhost/AbstractVirtualHost.java | 46 ++++++
.../virtualhost/ConnectionPrincipalStatistics.java | 13 +-
.../ConnectionPrincipalStatisticsCheckingTask.java | 42 +++++
.../ConnectionPrincipalStatisticsRegistry.java | 15 +-
.../ConnectionStatisticsRegistrySettings.java | 13 +-
.../virtualhost/QueueManagingVirtualHost.java | 5 +
.../ConnectionPrincipalStatisticsImpl.java | 83 ++++++++++
.../ConnectionPrincipalStatisticsRegistryImpl.java | 182 +++++++++++++++++++++
.../apache/qpid/server/model/VirtualHostTest.java | 130 ++++++++++-----
...nectionPrincipalStatisticsCheckingTaskTest.java | 47 ++++++
.../ConnectionPrincipalStatisticsImplTest.java | 100 +++++++++++
...nectionPrincipalStatisticsRegistryImplTest.java | 136 +++++++++++++++
.../server/security/access/config/AclAction.java | 36 ++--
.../security/access/config/AclRulePredicates.java | 130 +++++++++++----
.../security/access/config/ClientAction.java | 22 +--
.../FirewallRule.java => config/DynamicRule.java} | 31 +++-
.../security/access/config/ObjectProperties.java | 4 +-
.../access/config/RuleBasedAccessControl.java | 26 +--
.../server/security/access/config/RuleSet.java | 18 +-
.../ConnectionPrincipalFrequencyLimitRule.java} | 25 ++-
.../connection/ConnectionPrincipalLimitRule.java} | 24 ++-
.../ConnectionPrincipalStatisticsRule.java | 70 ++++++++
.../security/access/firewall/FirewallRule.java | 37 ++++-
.../access/firewall/HostnameFirewallRule.java | 20 +--
.../access/firewall/NetworkFirewallRule.java | 20 +--
.../security/access/config/AclActionTest.java | 2 +-
.../access/config/AclRulePredicatesTest.java | 19 +++
.../security/access/config/ClientActionTest.java | 88 ++++++++--
.../security/access/config/DynamicRuleTest.java | 57 +++++++
.../access/config/RuleBasedAccessControlTest.java | 5 +-
.../ConnectionPrincipalFrequencyLimitRuleTest.java | 84 ++++++++++
.../ConnectionPrincipalLimitRuleTest.java | 84 ++++++++++
.../server/protocol/v0_10/ServerSessionTest.java | 11 +-
...Java-Broker-Security-AccessControlProviders.xml | 54 +++++-
.../jms_1_1/extensions/acl/MessagingACLTest.java | 139 ++++++++++++++--
39 files changed, 1620 insertions(+), 279 deletions(-)
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/security/QpidPrincipal.java
b/broker-core/src/main/java/org/apache/qpid/server/security/QpidPrincipal.java
index 382fea8..3393098 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/security/QpidPrincipal.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/security/QpidPrincipal.java
@@ -55,8 +55,8 @@ public interface QpidPrincipal extends Principal, Serializable
"Can't find single %s in the authenticated
subject. There were %d "
+ "%s principals out of a total number of
principals of: %s",
principalClazz.getSimpleName(),
- principalClazz.getSimpleName(),
numberOfAuthenticatedPrincipals,
+ principalClazz.getSimpleName(),
authSubject.getPrincipals()));
}
return principals.iterator().next();
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/security/auth/manager/SpnegoAuthenticator.java
b/broker-core/src/main/java/org/apache/qpid/server/security/auth/manager/SpnegoAuthenticator.java
index 48cf4f9..70db671 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/security/auth/manager/SpnegoAuthenticator.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/security/auth/manager/SpnegoAuthenticator.java
@@ -215,11 +215,15 @@ public class SpnegoAuthenticator
{
TokenCarryingPrincipal principal = new TokenCarryingPrincipal()
{
+
+ private Map<String, String> _tokens =
Collections.singletonMap(RESPONSE_AUTH_HEADER_NAME,
+
NEGOTIATE_PREFIX + Base64.getEncoder()
+
.encodeToString(outToken));
+
@Override
public Map<String, String> getTokens()
{
- return
Collections.singletonMap(RESPONSE_AUTH_HEADER_NAME,
- NEGOTIATE_PREFIX +
Base64.getEncoder().encodeToString(outToken));
+ return _tokens;
}
@Override
@@ -233,6 +237,42 @@ public class SpnegoAuthenticator
{
return principalName;
}
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (!(o instanceof TokenCarryingPrincipal))
+ {
+ return false;
+ }
+
+ final TokenCarryingPrincipal that =
(TokenCarryingPrincipal) o;
+
+ if (!getName().equals(that.getName()))
+ {
+ return false;
+ }
+
+ if (!getTokens().equals(that.getTokens()))
+ {
+ return false;
+ }
+ return getOrigin() != null ?
getOrigin().equals(that.getOrigin()) : that.getOrigin() == null;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = getName().hashCode();
+ result = 31 * result + (getOrigin() != null ?
getOrigin().hashCode() : 0);
+ result = 31 * result + getTokens().hashCode();
+ return result;
+ }
+
};
return new AuthenticationResult(principal);
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
index e99c254..ebcd737 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
@@ -39,6 +39,7 @@ import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.Deletable;
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
public interface AMQPConnection<C extends AMQPConnection<C>>
extends Connection<C>, Deletable<C>, EventLoggerProvider
@@ -147,4 +148,10 @@ public interface AMQPConnection<C extends
AMQPConnection<C>>
@Override
AmqpPort<?> getPort();
+ void registered(ConnectionPrincipalStatistics
connectionPrincipalStatistics);
+
+ int getAuthenticatedPrincipalConnectionCount();
+
+ int getAuthenticatedPrincipalConnectionFrequency();
+
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
index ae4dee2..e45e993 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
@@ -83,6 +83,7 @@ import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.FixedKeyMapCreator;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,T>, T>
@@ -148,6 +149,7 @@ public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,
private long _maxUncommittedInMemorySize;
private final Map<ServerTransaction, Set<Ticker>> _transactionTickers =
new ConcurrentHashMap<>();
+ private volatile ConnectionPrincipalStatistics
_connectionPrincipalStatistics;
public AbstractAMQPConnection(Broker<?> broker,
ServerNetworkConnection network,
@@ -1157,4 +1159,30 @@ public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,
{
_localTransactionBegins.incrementAndGet();
}
+
+ @Override
+ public void registered(final ConnectionPrincipalStatistics
connectionPrincipalStatistics)
+ {
+ _connectionPrincipalStatistics = connectionPrincipalStatistics;
+ }
+
+ @Override
+ public int getAuthenticatedPrincipalConnectionCount()
+ {
+ if (_connectionPrincipalStatistics == null)
+ {
+ return 0;
+ }
+ return _connectionPrincipalStatistics.getConnectionCount();
+ }
+
+ @Override
+ public int getAuthenticatedPrincipalConnectionFrequency()
+ {
+ if (_connectionPrincipalStatistics == null)
+ {
+ return 0;
+ }
+ return _connectionPrincipalStatistics.getConnectionFrequency();
+ }
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index 3ddd26a..699f3c0 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -38,6 +38,7 @@ import java.nio.charset.StandardCharsets;
import java.security.AccessControlContext;
import java.security.Principal;
import java.security.PrivilegedAction;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -147,6 +148,7 @@ import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.HousekeepingExecutor;
import org.apache.qpid.server.util.Strings;
+import
org.apache.qpid.server.virtualhost.connection.ConnectionPrincipalStatisticsRegistryImpl;
public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>>
extends AbstractConfiguredObject<X>
implements QueueManagingVirtualHost<X>
@@ -282,6 +284,8 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
private PreferenceStore _preferenceStore;
private long _flowToDiskCheckPeriod;
private volatile boolean _isDiscardGlobalSharedSubscriptionLinksOnDetach;
+ private volatile ConnectionPrincipalStatisticsRegistry
_connectionPrincipalStatisticsRegistry;
+ private volatile HouseKeepingTask _statisticsCheckTask;
public AbstractVirtualHost(final Map<String, Object> attributes,
VirtualHostNode<?> virtualHostNode)
{
@@ -2441,6 +2445,7 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
@Override
public void run()
{
+ resetConnectionPrincipalStatisticsRegistry();
shutdownHouseKeeping();
closeNetworkConnectionScheduler();
if (_linkRegistry != null)
@@ -2596,6 +2601,9 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
{
if
(connectionEstablishmentPolicy.mayEstablishNewConnection(_connections,
connection))
{
+ final ConnectionPrincipalStatistics cps =
+
_connectionPrincipalStatisticsRegistry.connectionOpened(connection);
+ connection.registered(cps);
_connections.add(connection);
_totalConnectionCount.incrementAndGet();
@@ -2658,6 +2666,7 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
{
connection.popScheduler();
_connections.remove(connection);
+
_connectionPrincipalStatisticsRegistry.connectionClosed(connection);
return Futures.immediateFuture(null);
}
@@ -2701,6 +2710,7 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
updateAccessControl();
initialiseStatisticsReporting();
+ initialiseConnectionPrincipalStatisticsRegistry();
MessageStore messageStore = getMessageStore();
messageStore.openMessageStore(this);
@@ -2749,6 +2759,42 @@ public abstract class AbstractVirtualHost<X extends
AbstractVirtualHost<X>> exte
}
}
+ private void initialiseConnectionPrincipalStatisticsRegistry()
+ {
+ final long connectionFrequencyPeriodMillis =
getContextValue(Long.class, CONNECTION_FREQUENCY_PERIOD);
+ final Duration connectionFrequencyPeriod =
Duration.ofMillis(connectionFrequencyPeriodMillis);
+ final ConnectionPrincipalStatisticsRegistryImpl
connectionStatisticsRegistry =
+ new ConnectionPrincipalStatisticsRegistryImpl(() ->
connectionFrequencyPeriod);
+ HouseKeepingTask task = null;
+ long taskRunPeriod = connectionFrequencyPeriodMillis / 2;
+ if (taskRunPeriod > 0)
+ {
+ final AccessControlContext context =
+
getSystemTaskControllerContext("ConnectionPrincipalStatisticsCheck",
_principal);
+ task = new ConnectionPrincipalStatisticsCheckingTask(this,
context, connectionStatisticsRegistry);
+ scheduleHouseKeepingTask(taskRunPeriod, task);
+ }
+ _statisticsCheckTask = task;
+ _connectionPrincipalStatisticsRegistry = connectionStatisticsRegistry;
+ }
+
+ private void resetConnectionPrincipalStatisticsRegistry()
+ {
+ final HouseKeepingTask previousStatisticsCheckTask =
_statisticsCheckTask;
+ if (previousStatisticsCheckTask != null)
+ {
+ previousStatisticsCheckTask.cancel();
+ }
+ _statisticsCheckTask = null;
+ final ConnectionPrincipalStatisticsRegistry
connectionPrincipalStatisticsRegistry =
+ _connectionPrincipalStatisticsRegistry;
+ if (connectionPrincipalStatisticsRegistry != null)
+ {
+ connectionPrincipalStatisticsRegistry.reset();
+ }
+ _connectionPrincipalStatisticsRegistry = null;
+ }
+
private void postCreateDefaultExchangeTasks()
{
if(getContextValue(Boolean.class, USE_ASYNC_RECOVERY))
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatistics.java
similarity index 71%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatistics.java
index 72d8d69..661b89b 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatistics.java
@@ -16,17 +16,12 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.virtualhost;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
-
-public interface FirewallRule
+public interface ConnectionPrincipalStatistics
{
- boolean matches(InetAddress addressOfClient);
-
- ObjectProperties.Property getPropertyName();
+ int getConnectionCount();
- String getPropertyValue();
+ int getConnectionFrequency();
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTask.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTask.java
new file mode 100644
index 0000000..85322bd
--- /dev/null
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTask.java
@@ -0,0 +1,42 @@
+/*
+ * 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.virtualhost;
+
+import java.security.AccessControlContext;
+
+public class ConnectionPrincipalStatisticsCheckingTask extends HouseKeepingTask
+{
+ private static final String TASK_NAME =
"ConnectionPrincipalStatisticsCheck";
+ private final ConnectionPrincipalStatisticsRegistry
_connectionPrincipalStatisticsRegistry;
+
+ public ConnectionPrincipalStatisticsCheckingTask(final
QueueManagingVirtualHost virtualHost,
+ final
AccessControlContext controlContext,
+ final
ConnectionPrincipalStatisticsRegistry connectionPrincipalStatisticsRegistry)
+ {
+ super(TASK_NAME, virtualHost, controlContext);
+ _connectionPrincipalStatisticsRegistry =
connectionPrincipalStatisticsRegistry;
+ }
+
+ @Override
+ public void execute()
+ {
+
_connectionPrincipalStatisticsRegistry.reevaluateConnectionStatistics();
+ }
+}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsRegistry.java
similarity index 68%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsRegistry.java
index 72d8d69..2f0afba 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsRegistry.java
@@ -16,17 +16,18 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.virtualhost;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
+import org.apache.qpid.server.transport.AMQPConnection;
-public interface FirewallRule
+public interface ConnectionPrincipalStatisticsRegistry
{
- boolean matches(InetAddress addressOfClient);
+ ConnectionPrincipalStatistics connectionOpened(AMQPConnection<?>
connection);
- ObjectProperties.Property getPropertyName();
+ ConnectionPrincipalStatistics connectionClosed(AMQPConnection<?>
connection);
- String getPropertyValue();
+ void reevaluateConnectionStatistics();
+
+ void reset();
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionStatisticsRegistrySettings.java
similarity index 71%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionStatisticsRegistrySettings.java
index 72d8d69..d9611e6 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ConnectionStatisticsRegistrySettings.java
@@ -16,17 +16,12 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.virtualhost;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
+import java.time.Duration;
-public interface FirewallRule
+public interface ConnectionStatisticsRegistrySettings
{
- boolean matches(InetAddress addressOfClient);
-
- ObjectProperties.Property getPropertyName();
-
- String getPropertyValue();
+ Duration getConnectionFrequencyPeriod();
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
index 9f1b318..f76835d 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
@@ -104,6 +104,11 @@ public interface QueueManagingVirtualHost<X extends
QueueManagingVirtualHost<X>>
+ " link detaches. This is to avoid
leaking links with the Qpid JMS client.")
boolean DEFAULT_DISCARD_GLOBAL_SHARED_SUBSCRIPTION_LINKS_ON_DETACH = true;
+ String CONNECTION_FREQUENCY_PERIOD =
"qpid.virtualhost.connectionFrequencyPeriodInMillis";
+ @ManagedContextDefault(name = CONNECTION_FREQUENCY_PERIOD, description =
"Interval (in milliseconds) to evaluate connection frequency")
+ @SuppressWarnings("unused")
+ long DEFAULT_CONNECTION_FREQUENCY_PERIOD = 60 * 1000;
+
@ManagedAttribute( defaultValue = "${" +
VIRTUALHOST_STATISTICS_REPORING_PERIOD + "}", description = "Period (in
seconds) of the statistic report.")
int getStatisticsReportingPeriod();
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImpl.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImpl.java
new file mode 100644
index 0000000..9acd7a5
--- /dev/null
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImpl.java
@@ -0,0 +1,83 @@
+/*
+ * 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.virtualhost.connection;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
+
+class ConnectionPrincipalStatisticsImpl implements
ConnectionPrincipalStatistics
+{
+ private final int _connectionCount;
+ private final List<Long> _latestConnectionCreatedTimes;
+
+ ConnectionPrincipalStatisticsImpl(final int connectionCount, final
List<Long> latestConnectionCreatedTimes)
+ {
+ _connectionCount = connectionCount;
+ _latestConnectionCreatedTimes =
Collections.unmodifiableList(latestConnectionCreatedTimes);
+ }
+
+ @Override
+ public int getConnectionCount()
+ {
+ return _connectionCount;
+ }
+
+ @Override
+ public int getConnectionFrequency()
+ {
+ return _latestConnectionCreatedTimes.size();
+ }
+
+ List<Long> getLatestConnectionCreatedTimes()
+ {
+ return _latestConnectionCreatedTimes;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final ConnectionPrincipalStatisticsImpl that =
(ConnectionPrincipalStatisticsImpl) o;
+
+ if (_connectionCount != that._connectionCount)
+ {
+ return false;
+ }
+ return
_latestConnectionCreatedTimes.equals(that._latestConnectionCreatedTimes);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = _connectionCount;
+ result = 31 * result + _latestConnectionCreatedTimes.hashCode();
+ return result;
+ }
+}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImpl.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImpl.java
new file mode 100644
index 0000000..59d03ea
--- /dev/null
+++
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImpl.java
@@ -0,0 +1,182 @@
+/*
+ * 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.virtualhost.connection;
+
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import javax.security.auth.Subject;
+
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
+import
org.apache.qpid.server.virtualhost.ConnectionPrincipalStatisticsRegistry;
+import org.apache.qpid.server.virtualhost.ConnectionStatisticsRegistrySettings;
+
+public class ConnectionPrincipalStatisticsRegistryImpl implements
ConnectionPrincipalStatisticsRegistry
+{
+ private final Map<Principal, ConnectionPrincipalStatisticsImpl>
_principalStatistics = new ConcurrentHashMap<>();
+ private final ConnectionStatisticsRegistrySettings _settings;
+
+ public ConnectionPrincipalStatisticsRegistryImpl(final
ConnectionStatisticsRegistrySettings settings)
+ {
+ _settings = settings;
+ }
+
+ @Override
+ public ConnectionPrincipalStatistics connectionOpened(final
AMQPConnection<?> connection)
+ {
+ final Subject subject = connection.getSubject();
+ final AuthenticatedPrincipal principal =
AuthenticatedPrincipal.getAuthenticatedPrincipalFromSubject(subject);
+ return _principalStatistics.compute(principal,
+ (p, s) -> connectionOpened(s,
connection.getCreatedTime()));
+ }
+
+ @Override
+ public ConnectionPrincipalStatistics connectionClosed(final
AMQPConnection<?> connection)
+ {
+ final Subject subject = connection.getSubject();
+ final AuthenticatedPrincipal principal =
AuthenticatedPrincipal.getAuthenticatedPrincipalFromSubject(subject);
+ return _principalStatistics.computeIfPresent(principal, (p, s) ->
connectionClosed(s));
+ }
+
+ @Override
+ public void reevaluateConnectionStatistics()
+ {
+ new
HashSet<>(_principalStatistics.keySet()).forEach(this::reevaluateConnectionStatistics);
+ }
+
+ @Override
+ public void reset()
+ {
+ _principalStatistics.clear();
+ }
+
+ int getConnectionCount(final Principal principal)
+ {
+ ConnectionPrincipalStatistics cs = _principalStatistics.get(principal);
+ if (cs != null)
+ {
+ return cs.getConnectionCount();
+ }
+ return 0;
+ }
+
+ int getConnectionFrequency(final Principal principal)
+ {
+ ConnectionPrincipalStatistics cs = _principalStatistics.get(principal);
+ if (cs != null)
+ {
+ return cs.getConnectionFrequency();
+ }
+ return 0;
+ }
+
+ private ConnectionPrincipalStatisticsImpl connectionOpened(final
ConnectionPrincipalStatisticsImpl current,
+ final Date
createdTime)
+ {
+ if (current == null)
+ {
+ return new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(createdTime.getTime()));
+ }
+ else
+ {
+ final long frequencyPeriod = getConnectionFrequencyPeriodMillis();
+ final List<Long> connectionCreatedTimes;
+ if (frequencyPeriod > 0)
+ {
+ connectionCreatedTimes =
findTimesWithinPeriod(current.getLatestConnectionCreatedTimes(),
frequencyPeriod);
+ connectionCreatedTimes.add(createdTime.getTime());
+ }
+ else
+ {
+ connectionCreatedTimes = Collections.emptyList();
+ }
+ return new
ConnectionPrincipalStatisticsImpl(current.getConnectionCount() + 1,
connectionCreatedTimes);
+ }
+ }
+
+ private ConnectionPrincipalStatisticsImpl connectionClosed(final
ConnectionPrincipalStatisticsImpl current)
+ {
+ return createStatisticsOrNull(Math.max(0, current.getConnectionCount()
- 1), current.getLatestConnectionCreatedTimes());
+ }
+
+ private void reevaluateConnectionStatistics(final Principal
authenticatedPrincipal)
+ {
+ _principalStatistics.computeIfPresent(authenticatedPrincipal,
(principal, current) -> reevaluate(current));
+ }
+
+ private ConnectionPrincipalStatisticsImpl reevaluate(final
ConnectionPrincipalStatisticsImpl current)
+ {
+ return createStatisticsOrNull(current.getConnectionCount(),
current.getLatestConnectionCreatedTimes());
+ }
+
+ private ConnectionPrincipalStatisticsImpl createStatisticsOrNull(final int
openConnectionCount,
+ final
List<Long> currentConnectionCreatedTimes)
+ {
+ final List<Long> connectionCreatedTimes =
findTimesWithinPeriod(currentConnectionCreatedTimes,
getConnectionFrequencyPeriodMillis());
+ if (openConnectionCount == 0 && connectionCreatedTimes.isEmpty())
+ {
+ return null;
+ }
+ return new ConnectionPrincipalStatisticsImpl(openConnectionCount,
connectionCreatedTimes);
+ }
+
+ private List<Long> findTimesWithinPeriod(final Collection<Long>
currentConnectionCreatedTimes,
+ final long frequencyPeriod)
+ {
+
+ final List<Long> connectionCreatedTimes;
+ if (frequencyPeriod > 0)
+ {
+ final long periodEnd = System.currentTimeMillis();
+ final long periodStart = periodEnd - frequencyPeriod;
+ connectionCreatedTimes =
findTimesBetween(currentConnectionCreatedTimes, periodStart, periodEnd);
+ }
+ else
+ {
+ connectionCreatedTimes = Collections.emptyList();
+ }
+ return connectionCreatedTimes;
+ }
+
+ private List<Long> findTimesBetween(final Collection<Long>
connectionCreatedTimes,
+ final long periodStart,
+ final long periodEnd)
+ {
+ return connectionCreatedTimes.stream().mapToLong(Long::longValue)
+ .filter(t -> t >= periodStart && t <=
periodEnd)
+ .boxed()
+
.collect(Collectors.toCollection(ArrayList::new));
+ }
+
+ private long getConnectionFrequencyPeriodMillis()
+ {
+ return _settings.getConnectionFrequencyPeriod().toMillis();
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
b/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
index dc96a93..b3dea1b 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
@@ -26,7 +26,6 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
@@ -43,18 +42,20 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import javax.security.auth.Subject;
+
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -67,6 +68,7 @@ import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.security.AccessControl;
import org.apache.qpid.server.security.Result;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.store.ConfiguredObjectRecord;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.store.Event;
@@ -75,8 +77,7 @@ import
org.apache.qpid.server.store.handler.ConfiguredObjectRecordHandler;
import org.apache.qpid.server.store.preferences.PreferenceStore;
import org.apache.qpid.server.store.preferences.PreferenceStoreUpdater;
import org.apache.qpid.server.transport.AMQPConnection;
-import org.apache.qpid.server.transport.AbstractAMQPConnection;
-import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
import org.apache.qpid.server.virtualhost.NodeAutoCreationPolicy;
import org.apache.qpid.server.virtualhost.NoopConnectionEstablishmentPolicy;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -333,13 +334,11 @@ public class VirtualHostTest extends UnitTestBase
QueueManagingVirtualHost<?> virtualHost =
createVirtualHost(virtualHostName);
assertEquals("Unexpected state", State.ACTIVE, virtualHost.getState());
- AbstractAMQPConnection connection =
createMockProtocolConnection(virtualHost);
assertEquals("Unexpected number of connections before connection
registered",
(long) 0,
virtualHost.getConnectionCount());
- AMQPConnection modelConnection = mock(AMQPConnection.class);
-
when(modelConnection.closeAsync()).thenReturn(Futures.immediateFuture(null));
+ AMQPConnection modelConnection = getMockConnection();
virtualHost.registerConnection(modelConnection, new
NoopConnectionEstablishmentPolicy());
assertEquals("Unexpected number of connections after connection
registered",
@@ -364,13 +363,11 @@ public class VirtualHostTest extends UnitTestBase
QueueManagingVirtualHost<?> virtualHost =
createVirtualHost(virtualHostName);
assertEquals("Unexpected state", State.ACTIVE, virtualHost.getState());
- AbstractAMQPConnection connection =
createMockProtocolConnection(virtualHost);
assertEquals("Unexpected number of connections before connection
registered",
(long) 0,
virtualHost.getConnectionCount());
- AMQPConnection modelConnection = mock(AMQPConnection.class);
-
when(modelConnection.closeAsync()).thenReturn(Futures.immediateFuture(null));
+ AMQPConnection modelConnection = getMockConnection();
virtualHost.registerConnection(modelConnection, new
NoopConnectionEstablishmentPolicy());
assertEquals("Unexpected number of connections after connection
registered",
@@ -502,9 +499,9 @@ public class VirtualHostTest extends UnitTestBase
public void testExistingConnectionBlocking()
{
VirtualHost<?> host = createVirtualHost(getTestName());
- AbstractAMQPConnection connection = mock(AbstractAMQPConnection.class);
+ AMQPConnection connection = getMockConnection();
host.registerConnection(connection, new
NoopConnectionEstablishmentPolicy());
- ((EventListener)host).event(Event.PERSISTENT_MESSAGE_SIZE_OVERFULL);
+ ((EventListener) host).event(Event.PERSISTENT_MESSAGE_SIZE_OVERFULL);
verify(connection).block();
}
@@ -590,6 +587,46 @@ public class VirtualHostTest extends UnitTestBase
}
@Test
+ public void
testRegisterConnectionCausesUpdateOfAuthenticatedPrincipalConnectionCountAndFrequency()
+ {
+ final NoopConnectionEstablishmentPolicy policy = new
NoopConnectionEstablishmentPolicy();
+ final QueueManagingVirtualHost<?> vhost =
createVirtualHost(getTestName());
+ final Principal principal = mockAuthenticatedPrincipal(getTestName());
+ final Principal principal2 = mockAuthenticatedPrincipal(getTestName()
+ "_2");
+ final AMQPConnection<?> connection1 = mockAmqpConnection(principal);
+ final AMQPConnection<?> connection2 = mockAmqpConnection(principal);
+ final AMQPConnection<?> connection3 = mockAmqpConnection(principal2);
+
+ vhost.registerConnection(connection1, policy);
+ verify(connection1).registered(argThat(new
ConnectionPrincipalStatisticsArgumentMatcher(1, 1)));
+
+ vhost.registerConnection(connection2, policy);
+ verify(connection2).registered(argThat(new
ConnectionPrincipalStatisticsArgumentMatcher(2, 2)));
+
+ vhost.registerConnection(connection3, policy);
+ verify(connection3).registered(argThat(new
ConnectionPrincipalStatisticsArgumentMatcher(1, 1)));
+ }
+
+ @Test
+ public void
testDeregisterConnectionAffectsAuthenticatedPrincipalConnectionCountAndFrequency()
+ {
+ final Principal principal = mockAuthenticatedPrincipal(getTestName());
+ final NoopConnectionEstablishmentPolicy policy = new
NoopConnectionEstablishmentPolicy();
+ final QueueManagingVirtualHost<?> vhost =
createVirtualHost(getTestName());
+
+ final AMQPConnection<?> connection1 = mockAmqpConnection(principal);
+ final AMQPConnection<?> connection2 = mockAmqpConnection(principal);
+
+ vhost.registerConnection(connection1, policy);
+ verify(connection1).registered(argThat(new
ConnectionPrincipalStatisticsArgumentMatcher(1, 1)));
+
+ vhost.deregisterConnection(connection1);
+ vhost.registerConnection(connection2, policy);
+
+ verify(connection2).registered(argThat(new
ConnectionPrincipalStatisticsArgumentMatcher(1, 2)));
+ }
+
+ @Test
public void testStopVirtualhostClosesConnections()
{
QueueManagingVirtualHost<?> vhost = createVirtualHost("sdf");
@@ -775,12 +812,29 @@ public class VirtualHostTest extends UnitTestBase
private AMQPConnection<?> getMockConnection()
{
+ return mockAmqpConnection(mockAuthenticatedPrincipal(getTestName()));
+ }
+
+ private AMQPConnection<?> mockAmqpConnection(final Principal principal)
+ {
AMQPConnection<?> connection = mock(AMQPConnection.class);
+ when(connection.getAuthorizedPrincipal()).thenReturn(principal);
+ final Subject subject =
+ new Subject(true, Collections.singleton(principal),
Collections.emptySet(), Collections.emptySet());
+ when(connection.getSubject()).thenReturn(subject);
final ListenableFuture<Void> listenableFuture =
Futures.immediateFuture(null);
when(connection.closeAsync()).thenReturn(listenableFuture);
+ when(connection.getCreatedTime()).thenReturn(new Date());
return connection;
}
+ private Principal mockAuthenticatedPrincipal(final String principalName)
+ {
+ final Principal principal = mock(Principal.class);
+ when(principal.getName()).thenReturn(principalName);
+ return new AuthenticatedPrincipal(principal);
+ }
+
private QueueManagingVirtualHost<?> createVirtualHost(final String
virtualHostName)
{
return createVirtualHost(virtualHostName, Collections.<String,
Object>emptyMap());
@@ -803,37 +857,6 @@ public class VirtualHostTest extends UnitTestBase
return host;
}
- private AbstractAMQPConnection createMockProtocolConnection(final
VirtualHost<?> virtualHost)
- {
- final AbstractAMQPConnection connection =
mock(AbstractAMQPConnection.class);
- final List<Action<?>> tasks = new ArrayList<>();
- final ArgumentCaptor<Action> deleteTaskCaptor =
ArgumentCaptor.forClass(Action.class);
- Answer answer = new Answer()
- {
- @Override
- public Object answer(final InvocationOnMock invocation)
- {
- return tasks.add(deleteTaskCaptor.getValue());
- }
- };
-
doAnswer(answer).when(connection).addDeleteTask(deleteTaskCaptor.capture());
- when(connection.getAddressSpace()).thenReturn(virtualHost);
- doAnswer(new Answer()
- {
- @Override
- public Object answer(final InvocationOnMock invocation)
- {
- for(Action action : tasks)
- {
- action.performAction(connection);
- }
- return null;
- }
-
}).when(connection).sendConnectionCloseAsync(any(AMQPConnection.CloseReason.class),
anyString());
- when(connection.getRemoteAddressString()).thenReturn("peer:1234");
- return connection;
- }
-
private static ConfiguredObjectRecord matchesRecord(UUID id, String type)
{
return argThat(new MinimalConfiguredObjectRecordMatcher(id, type));
@@ -856,4 +879,25 @@ public class VirtualHostTest extends UnitTestBase
return (_id.equals(rhs.getId()) || _type.equals(rhs.getType()));
}
}
+
+ private static class ConnectionPrincipalStatisticsArgumentMatcher
implements ArgumentMatcher<ConnectionPrincipalStatistics>
+ {
+
+ private final int _expectedConnectionCount;
+ private final int _expectedConnectionFrequency;
+
+ ConnectionPrincipalStatisticsArgumentMatcher(final int
expectedConnectionCount,
+ final int
expectedConnectionFrequency)
+ {
+ _expectedConnectionCount = expectedConnectionCount;
+ _expectedConnectionFrequency = expectedConnectionFrequency;
+ }
+
+ @Override
+ public boolean matches(final ConnectionPrincipalStatistics
connectionPrincipalStatistics)
+ {
+ return connectionPrincipalStatistics.getConnectionFrequency() ==
_expectedConnectionFrequency
+ && connectionPrincipalStatistics.getConnectionCount() ==
_expectedConnectionCount;
+ }
+ }
}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTaskTest.java
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTaskTest.java
new file mode 100644
index 0000000..1764050
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/ConnectionPrincipalStatisticsCheckingTaskTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.virtualhost;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.security.AccessController;
+
+import org.junit.Test;
+
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class ConnectionPrincipalStatisticsCheckingTaskTest extends UnitTestBase
+{
+ @Test
+ public void execute()
+ {
+ final QueueManagingVirtualHost vh =
mock(QueueManagingVirtualHost.class);
+ when(vh.getName()).thenReturn(getTestName());
+ final ConnectionPrincipalStatisticsRegistry registry =
mock(ConnectionPrincipalStatisticsRegistry.class);
+ ConnectionPrincipalStatisticsCheckingTask task =
+ new ConnectionPrincipalStatisticsCheckingTask(vh,
AccessController.getContext(), registry);
+
+ task.execute();
+
+ verify(registry).reevaluateConnectionStatistics();
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImplTest.java
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImplTest.java
new file mode 100644
index 0000000..fd08684
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsImplTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.virtualhost.connection;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.qpid.server.virtualhost.ConnectionPrincipalStatistics;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class ConnectionPrincipalStatisticsImplTest extends UnitTestBase
+{
+
+ @Test
+ public void getOpenConnectionCount()
+ {
+ final ConnectionPrincipalStatistics stats =
+ new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(System.currentTimeMillis()));
+
+ assertEquals(1, stats.getConnectionCount());
+ }
+
+ @Test
+ public void getOpenConnectionFrequency()
+ {
+ final long connectionCreatedTime = System.currentTimeMillis();
+ final ConnectionPrincipalStatistics stats =
+ new ConnectionPrincipalStatisticsImpl(1,
+
Arrays.asList(connectionCreatedTime - 1000, connectionCreatedTime));
+ assertEquals(2, stats.getConnectionFrequency());
+ }
+
+ @Test
+ public void getLatestConnectionCreatedTimes()
+ {
+ final long connectionCreatedTime = System.currentTimeMillis();
+ final List<Long> connectionCreatedTimes =
Arrays.asList(connectionCreatedTime - 1000, connectionCreatedTime);
+ final ConnectionPrincipalStatisticsImpl stats = new
ConnectionPrincipalStatisticsImpl(1, connectionCreatedTimes);
+ assertEquals(connectionCreatedTimes,
stats.getLatestConnectionCreatedTimes());
+ }
+
+ @Test
+ public void equals()
+ {
+ final long connectionCreatedTime = System.currentTimeMillis();
+ final ConnectionPrincipalStatistics stats1 =
+ new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(connectionCreatedTime));
+ final ConnectionPrincipalStatistics stats2 =
+ new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(connectionCreatedTime));
+ assertEquals(stats1, stats2);
+
+ final long connectionCreatedTime2 = System.currentTimeMillis();
+ final ConnectionPrincipalStatistics stats3 =
+ new ConnectionPrincipalStatisticsImpl(2,
Arrays.asList(connectionCreatedTime, connectionCreatedTime2));
+
+ assertNotEquals(stats2, stats3);
+ assertNotEquals(stats1, stats3);
+ }
+
+ @Test
+ public void testHashCode()
+ {
+ final long connectionCreatedTime = System.currentTimeMillis();
+ final ConnectionPrincipalStatistics stats1 =
+ new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(connectionCreatedTime));
+ final ConnectionPrincipalStatistics stats2 =
+ new ConnectionPrincipalStatisticsImpl(1,
Collections.singletonList(connectionCreatedTime));
+ assertEquals(stats1.hashCode(), stats2.hashCode());
+
+ final long connectionCreatedTime2 = System.currentTimeMillis();
+ final ConnectionPrincipalStatistics stats3 =
+ new ConnectionPrincipalStatisticsImpl(2,
Arrays.asList(connectionCreatedTime, connectionCreatedTime2));
+
+ assertNotEquals(stats2.hashCode(), stats3.hashCode());
+ assertNotEquals(stats1.hashCode(), stats3.hashCode());
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImplTest.java
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImplTest.java
new file mode 100644
index 0000000..fa7b294
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/connection/ConnectionPrincipalStatisticsRegistryImplTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.virtualhost.connection;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.security.Principal;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Date;
+
+import javax.security.auth.Subject;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.virtualhost.ConnectionStatisticsRegistrySettings;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class ConnectionPrincipalStatisticsRegistryImplTest extends UnitTestBase
+{
+ private static final Duration CONNECTION_FREQUENCY_PERIOD =
Duration.ofMillis(5000);
+ private ConnectionPrincipalStatisticsRegistryImpl _statisticsRegistry;
+ private AuthenticatedPrincipal _authorizedPrincipal;
+ private ConnectionStatisticsRegistrySettings _settings;
+
+ @Before
+ public void setUp()
+ {
+ _settings = mock(ConnectionStatisticsRegistrySettings.class);
+
when(_settings.getConnectionFrequencyPeriod()).thenReturn(CONNECTION_FREQUENCY_PERIOD);
+ _statisticsRegistry = new
ConnectionPrincipalStatisticsRegistryImpl(_settings);
+ _authorizedPrincipal = new
AuthenticatedPrincipal(mock(Principal.class));
+ }
+
+ @Test
+ public void onConnectionOpen()
+ {
+ final AMQPConnection connection = mockConnection();
+
+ _statisticsRegistry.connectionOpened(connection);
+
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(1)));
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(1)));
+ }
+
+ @Test
+ public void onConnectionClose()
+ {
+ final AMQPConnection connection1 = mockConnection();
+
+ _statisticsRegistry.connectionOpened(connection1);
+ _statisticsRegistry.connectionClosed(connection1);
+
+ final AMQPConnection connection2 = mockConnection();
+ _statisticsRegistry.connectionOpened(connection2);
+
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(1)));
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(2)));
+ }
+
+ @Test
+ public void reevaluateConnectionPrincipalStatistics() throws
InterruptedException
+ {
+ final AMQPConnection connection1 = mockConnection();
+
+ _statisticsRegistry.connectionOpened(connection1);
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(1)));
+
+ _statisticsRegistry.reevaluateConnectionStatistics();
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(1)));
+
+
when(_settings.getConnectionFrequencyPeriod()).thenReturn(Duration.ofMillis(1));
+ Thread.sleep(_settings.getConnectionFrequencyPeriod().toMillis() + 1);
+
+ _statisticsRegistry.reevaluateConnectionStatistics();
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(1)));
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(0)));
+ }
+
+ @Test
+ public void getConnectionFrequencyAfterExpirationOfFrequencyPeriod()
throws InterruptedException
+ {
+ final AMQPConnection connection1 = mockConnection();
+ _statisticsRegistry.connectionOpened(connection1);
+
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(1)));
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(1)));
+
+
when(_settings.getConnectionFrequencyPeriod()).thenReturn(Duration.ofMillis(1));
+ Thread.sleep(_settings.getConnectionFrequencyPeriod().toMillis() + 1);
+
+ final AMQPConnection connection2 = mockConnection();
+ _statisticsRegistry.connectionOpened(connection2);
+
+
assertThat(_statisticsRegistry.getConnectionCount(_authorizedPrincipal),
is(equalTo(2)));
+
assertThat(_statisticsRegistry.getConnectionFrequency(_authorizedPrincipal),
is(equalTo(1)));
+ }
+
+ private AMQPConnection mockConnection()
+ {
+ final Subject subject = new Subject(true,
+
Collections.singleton(_authorizedPrincipal),
+ Collections.emptySet(),
+ Collections.emptySet());
+
+ final AMQPConnection connection = mock(AMQPConnection.class);
+
when(connection.getAuthorizedPrincipal()).thenReturn(_authorizedPrincipal);
+ when(connection.getSubject()).thenReturn(subject);
+ when(connection.getCreatedTime()).thenReturn(new Date());
+ return connection;
+ }
+}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclAction.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclAction.java
index 78866f5..89a09bc 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclAction.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclAction.java
@@ -18,21 +18,18 @@
*/
package org.apache.qpid.server.security.access.config;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
-
-import org.apache.qpid.server.security.access.firewall.FirewallRule;
+import java.util.Objects;
public class AclAction
{
private Action _action;
- private FirewallRule _firewallRule;
+ private AclRulePredicates _predicates;
public AclAction(LegacyOperation operation, ObjectType object,
AclRulePredicates predicates)
{
_action = new Action(operation, object,
predicates.getObjectProperties());
- _firewallRule = predicates.getFirewallRule();
+ _predicates = predicates;
}
public AclAction(LegacyOperation operation)
@@ -45,9 +42,9 @@ public class AclAction
_action = new Action(operation, object, properties);
}
- public FirewallRule getFirewallRule()
+ public DynamicRule getDynamicRule()
{
- return _firewallRule;
+ return _predicates == null ? null : _predicates.getDynamicRule();
}
public Action getAction()
@@ -78,9 +75,7 @@ public class AclAction
{
return false;
}
- return !(getFirewallRule() != null
- ? !getFirewallRule().equals(aclAction.getFirewallRule())
- : aclAction.getFirewallRule() != null);
+ return Objects.equals(_predicates, aclAction._predicates);
}
@@ -88,7 +83,7 @@ public class AclAction
public int hashCode()
{
int result = getAction() != null ? getAction().hashCode() : 0;
- result = 31 * result + (getFirewallRule() != null ?
getFirewallRule().hashCode() : 0);
+ result = 31 * result + (_predicates != null ? _predicates.hashCode() :
0);
return result;
}
@@ -97,24 +92,13 @@ public class AclAction
{
return "AclAction[" +
"action=" + _action +
- ", firewallRule=" + _firewallRule +
+ ", predicates=" + _predicates +
']';
}
public Map<ObjectProperties.Property, String> getAttributes()
{
- final ObjectProperties properties = _action.getProperties();
- final Map<ObjectProperties.Property, String> attributes = new
HashMap<>(properties.asPropertyMap());
- final Set<String> attributeNames = properties.getAttributeNames();
- if (attributeNames != null && !attributeNames.isEmpty())
- {
- attributes.put(ObjectProperties.Property.ATTRIBUTES,
String.join(",", attributeNames));
- }
- final FirewallRule firewallRule = getFirewallRule();
- if (firewallRule != null)
- {
- attributes.put(firewallRule.getPropertyName(),
firewallRule.getPropertyValue());
- }
- return attributes;
+ return _predicates.getParsedProperties();
}
+
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclRulePredicates.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclRulePredicates.java
index 5b5b73d..a172b01 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclRulePredicates.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/AclRulePredicates.java
@@ -18,19 +18,23 @@
*/
package org.apache.qpid.server.security.access.config;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
+import java.util.stream.Collectors;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.server.security.access.config.ObjectProperties.Property;
-import org.apache.qpid.server.security.access.firewall.FirewallRule;
+import
org.apache.qpid.server.security.access.config.connection.ConnectionPrincipalFrequencyLimitRule;
+import
org.apache.qpid.server.security.access.config.connection.ConnectionPrincipalLimitRule;
import org.apache.qpid.server.security.access.firewall.FirewallRuleFactory;
/**
* Represents the predicates on an ACL rule by combining predicates relating
to the object being operated on
- * (e.g. name=foo) with firewall rules.
+ * (e.g. name=foo) with dynamic rules.
*/
public class AclRulePredicates
{
@@ -38,11 +42,10 @@ public class AclRulePredicates
private static final String SEPARATOR = ",";
- private ObjectProperties _properties = new ObjectProperties();
-
- private FirewallRule _firewallRule;
-
- private FirewallRuleFactory _firewallRuleFactory = new
FirewallRuleFactory();
+ private final ObjectProperties _properties = new ObjectProperties();
+ private final Map<Property, String> _parsedProperties = new HashMap<>();
+ private volatile DynamicRule _dynamicRule = s -> true;
+ private volatile FirewallRuleFactory _firewallRuleFactory = new
FirewallRuleFactory();
public AclRulePredicates()
{
@@ -50,9 +53,9 @@ public class AclRulePredicates
public AclRulePredicates(Map<Property, String> values)
{
- if(values != null)
+ if (values != null)
{
- for(Map.Entry<Property, String> entry : values.entrySet())
+ for (Map.Entry<Property, String> entry : values.entrySet())
{
addPropertyValue(entry.getKey(), entry.getValue());
}
@@ -64,64 +67,129 @@ public class AclRulePredicates
ObjectProperties.Property property =
ObjectProperties.Property.parse(key);
addPropertyValue(property, value);
-
- LOGGER.debug("Parsed {} with value {}", property, value);
+ _parsedProperties.put(property, value);
+ LOGGER.debug("Parsed {} with value {}", property, value);
}
private void addPropertyValue(final Property property, final String value)
{
- if(property == Property.FROM_HOSTNAME)
+ final DynamicRule dynamicRule = _dynamicRule;
+ if (property == Property.FROM_HOSTNAME)
{
- checkFirewallRuleNotAlreadyDefined(property.name(), value);
- _firewallRule =
_firewallRuleFactory.createForHostname(value.split(SEPARATOR));
+ checkFirewallRuleNotAlreadyDefined(property, value,
Property.FROM_NETWORK);
+ _dynamicRule =
dynamicRule.and(_firewallRuleFactory.createForHostname(value.split(SEPARATOR)));
}
- else if(property == Property.FROM_NETWORK)
+ else if (property == Property.FROM_NETWORK)
{
- checkFirewallRuleNotAlreadyDefined(property.name(), value);
- _firewallRule =
_firewallRuleFactory.createForNetwork(value.split(SEPARATOR));
+ checkFirewallRuleNotAlreadyDefined(property, value,
Property.FROM_HOSTNAME);
+ _dynamicRule =
dynamicRule.and(_firewallRuleFactory.createForNetwork(value.split(SEPARATOR)));
}
else if (property == Property.ATTRIBUTES)
{
_properties.setAttributeNames(Sets.newHashSet((value.split(SEPARATOR))));
}
+ else if (property == Property.CONNECTION_LIMIT)
+ {
+ checkPropertyAlreadyDefined(property);
+ final int limit = getLimit(property, value);
+ _dynamicRule = dynamicRule.and(new
ConnectionPrincipalLimitRule(limit));
+ }
+ else if (property == Property.CONNECTION_FREQUENCY_LIMIT)
+ {
+ checkPropertyAlreadyDefined(property);
+ final int limit = getLimit(property, value);
+ _dynamicRule = dynamicRule.and(new
ConnectionPrincipalFrequencyLimitRule(limit));
+ }
else
{
_properties.put(property, value);
}
}
- private void checkFirewallRuleNotAlreadyDefined(String key, String value)
+ private int getLimit(final Property property, final String value)
+ {
+ int limit;
+ try
+ {
+ limit = Integer.parseInt(value);
+ }
+ catch (Exception e)
+ {
+ throw new IllegalStateException(String.format("Property '%s' value
'%s' is not integer", property, value));
+ }
+ return limit;
+ }
+
+ private void checkFirewallRuleNotAlreadyDefined(Property property, String
value, Property... exclusiveProperty)
+ {
+ checkPropertyAlreadyDefined(property);
+ for (Property p : exclusiveProperty)
+ {
+ if (_parsedProperties.containsKey(p))
+ {
+ throw new IllegalStateException(
+ String.format("Cannot parse '%s=%s' because property
'%s' has already been defined",
+ property.toString().toLowerCase(),
+ value,
+ p));
+ }
+ }
+ }
+
+ private void checkPropertyAlreadyDefined(Property property)
+ {
+ if (_parsedProperties.containsKey(property))
+ {
+ throw new IllegalStateException(String.format("Property '%s' has
already been defined", property.toString().toLowerCase()));
+ }
+ }
+
+ @Override
+ public boolean equals(final Object o)
{
- if(_firewallRule != null)
+ if (this == o)
{
- throw new IllegalStateException(
- "Cannot parse " + key + "=" + value
- + " because firewall rule " + _firewallRule + " has
already been defined");
+ return true;
}
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final AclRulePredicates that = (AclRulePredicates) o;
+
+ return _parsedProperties.equals(that._parsedProperties);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return _parsedProperties.hashCode();
}
@Override
public String toString()
{
- return "AclRulePredicates[" +
- "properties=" + _properties +
- ", firewallRule=" + _firewallRule +
- ']';
+ return String.format("AclRulePredicates[%s]",
+ _parsedProperties.entrySet()
+ .stream()
+ .map(e -> e.getKey() + "=" +
e.getValue())
+ .collect(Collectors.joining("
")));
}
- public FirewallRule getFirewallRule()
+ DynamicRule getDynamicRule()
{
- return _firewallRule;
+ return _dynamicRule;
}
- public ObjectProperties getObjectProperties()
+ ObjectProperties getObjectProperties()
{
return _properties;
}
- public ObjectProperties getFirewallProperties()
+ Map<Property, String> getParsedProperties()
{
- return _properties;
+ return Collections.unmodifiableMap(_parsedProperties);
}
void setFirewallRuleFactory(FirewallRuleFactory firewallRuleFactory)
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ClientAction.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ClientAction.java
index b18bd2a..23f0605 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ClientAction.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ClientAction.java
@@ -18,9 +18,7 @@
*/
package org.apache.qpid.server.security.access.config;
-import java.net.InetAddress;
-
-import org.apache.qpid.server.security.access.firewall.FirewallRule;
+import javax.security.auth.Subject;
/**
* I represent an {@link Action} taken by a client from a known address. The
address is used to
@@ -30,7 +28,7 @@ public class ClientAction
{
private Action _clientAction;
- public ClientAction(Action clientAction)
+ ClientAction(Action clientAction)
{
_clientAction = clientAction;
}
@@ -40,23 +38,15 @@ public class ClientAction
_clientAction = new Action(operation, objectType, properties);
}
- public boolean matches(AclAction ruleAction, InetAddress addressOfClient)
+ public boolean matches(AclAction ruleAction, final Subject subject)
{
return _clientAction.matches(ruleAction.getAction())
- && addressOfClientMatches(ruleAction, addressOfClient);
+ && dynamicMatches(ruleAction.getDynamicRule(), subject);
}
- private boolean addressOfClientMatches(AclAction ruleAction, InetAddress
addressOfClient)
+ private boolean dynamicMatches(final DynamicRule dynamicRule, final
Subject subject)
{
- FirewallRule firewallRule = ruleAction.getFirewallRule();
- if(firewallRule == null || addressOfClient == null)
- {
- return true;
- }
- else
- {
- return firewallRule.matches(addressOfClient);
- }
+ return dynamicRule == null || dynamicRule.matches(subject);
}
public LegacyOperation getOperation()
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/DynamicRule.java
similarity index 53%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/DynamicRule.java
index 72d8d69..ede7d13 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/DynamicRule.java
@@ -16,17 +16,34 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.security.access.config;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
+import java.util.Objects;
-public interface FirewallRule
+import javax.security.auth.Subject;
+
+public interface DynamicRule
{
- boolean matches(InetAddress addressOfClient);
+ boolean matches(Subject subject);
+
+ default DynamicRule and(DynamicRule other)
+ {
+ Objects.requireNonNull(other);
- ObjectProperties.Property getPropertyName();
+ return new DynamicRule()
+ {
+ @Override
+ public boolean matches(final Subject subject)
+ {
+ return DynamicRule.this.matches(subject) &&
other.matches(subject);
+ }
- String getPropertyValue();
+ @Override
+ public String toString()
+ {
+ return String.format("%s and %s", DynamicRule.this.toString(),
other.toString());
+ }
+ };
+ }
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ObjectProperties.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ObjectProperties.java
index deec6f7..a3bc747 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ObjectProperties.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/ObjectProperties.java
@@ -64,7 +64,9 @@ public class ObjectProperties
VIRTUALHOST_NAME,
METHOD_NAME,
ATTRIBUTES,
- CREATED_BY;
+ CREATED_BY,
+ CONNECTION_LIMIT,
+ CONNECTION_FREQUENCY_LIMIT;
private static final Map<String, Property> _canonicalNameToPropertyMap
= new HashMap<String, ObjectProperties.Property>();
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControl.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControl.java
index 44fa558..07cb206 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControl.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControl.java
@@ -20,21 +20,17 @@
*/
package org.apache.qpid.server.security.access.config;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
import java.security.AccessController;
import java.util.Collections;
import java.util.Map;
-import java.util.Set;
import javax.security.auth.Subject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.server.connection.ConnectionPrincipal;
-import org.apache.qpid.server.model.*;
+import org.apache.qpid.server.model.Model;
+import org.apache.qpid.server.model.PermissionedObject;
import org.apache.qpid.server.security.AccessControl;
import org.apache.qpid.server.security.Result;
import org.apache.qpid.server.security.access.Operation;
@@ -78,7 +74,6 @@ public class RuleBasedAccessControl implements
AccessControl<CachingSecurityToke
@Override
public Result authorise(LegacyOperation operation, ObjectType objectType,
ObjectProperties properties)
{
- InetAddress addressOfClient = null;
final Subject subject =
Subject.getSubject(AccessController.getContext());
// Abstain if there is no subject/principal associated with this thread
@@ -87,30 +82,19 @@ public class RuleBasedAccessControl implements
AccessControl<CachingSecurityToke
return Result.DEFER;
}
- Set<ConnectionPrincipal> principals =
subject.getPrincipals(ConnectionPrincipal.class);
- if(!principals.isEmpty())
- {
- SocketAddress address =
principals.iterator().next().getConnection().getRemoteSocketAddress();
- if(address instanceof InetSocketAddress)
- {
- addressOfClient = ((InetSocketAddress) address).getAddress();
- }
- }
if(LOGGER.isDebugEnabled())
{
- LOGGER.debug("Checking " + operation + " " + objectType + " " +
- (addressOfClient == null ? "" : addressOfClient));
+ LOGGER.debug("Checking " + operation + " " + objectType );
}
try
{
- return _ruleSet.check(subject, operation, objectType, properties,
addressOfClient);
+ return _ruleSet.check(subject, operation, objectType, properties);
}
catch(Exception e)
{
- LOGGER.error("Unable to check " + operation + " " + objectType + "
"
- + (addressOfClient == null ? "" : addressOfClient),
e);
+ LOGGER.error("Unable to check " + operation + " " + objectType ,
e);
return Result.DENIED;
}
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
index 151e1ac..6b3371c 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
@@ -18,7 +18,6 @@
*/
package org.apache.qpid.server.security.access.config;
-import java.net.InetAddress;
import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
@@ -123,16 +122,6 @@ public class RuleSet implements EventLoggerProvider
}
/**
- * Checks for the case when the client's address is not known.
- *
- * @see #check(Subject, LegacyOperation, ObjectType, ObjectProperties,
InetAddress)
- */
- public Result check(Subject subject, LegacyOperation operation, ObjectType
objectType, ObjectProperties properties)
- {
- return check(subject, operation, objectType, properties, null);
- }
-
- /**
* Check the authorisation granted to a particular identity for an
operation on an object type with
* specific properties.
*
@@ -141,7 +130,10 @@ public class RuleSet implements EventLoggerProvider
* the first match found, or denies access if there are no matching rules.
Normally, it would be expected
* to have a default deny or allow rule at the end of an access
configuration however.
*/
- public Result check(Subject subject, LegacyOperation operation, ObjectType
objectType, ObjectProperties properties, InetAddress addressOfClient)
+ public Result check(Subject subject,
+ LegacyOperation operation,
+ ObjectType objectType,
+ ObjectProperties properties)
{
ClientAction action = new ClientAction(operation, objectType,
properties);
@@ -186,7 +178,7 @@ public class RuleSet implements EventLoggerProvider
{
LOGGER.debug("Checking against rule: {}", rule);
- if (action.matches(rule.getAclAction(), addressOfClient))
+ if (action.matches(rule.getAclAction(), subject))
{
RuleOutcome ruleOutcome = rule.getRuleOutcome();
LOGGER.debug("Action matches. Result: {}", ruleOutcome);
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRule.java
similarity index 53%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRule.java
index 72d8d69..8f4e7e4 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRule.java
@@ -16,17 +16,30 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.security.access.config.connection;
import org.apache.qpid.server.security.access.config.ObjectProperties;
+import org.apache.qpid.server.transport.AMQPConnection;
-public interface FirewallRule
+public class ConnectionPrincipalFrequencyLimitRule extends
ConnectionPrincipalStatisticsRule
{
- boolean matches(InetAddress addressOfClient);
+ public ConnectionPrincipalFrequencyLimitRule(final int limit)
+ {
+ super(limit);
+ }
- ObjectProperties.Property getPropertyName();
+ @Override
+ boolean matches(final AMQPConnection<?> connection)
+ {
+ return connection.getAuthenticatedPrincipalConnectionFrequency() <=
getLimit();
+ }
- String getPropertyValue();
+ @Override
+ public String toString()
+ {
+ return String.format("ConnectionPrincipalFrequencyLimitRule{%s=%d}",
+
ObjectProperties.Property.CONNECTION_FREQUENCY_LIMIT.name().toLowerCase(),
+ getLimit());
+ }
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRule.java
similarity index 55%
copy from
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
copy to
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRule.java
index 72d8d69..94a93f5 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRule.java
@@ -16,17 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.qpid.server.security.access.firewall;
-import java.net.InetAddress;
+package org.apache.qpid.server.security.access.config.connection;
import org.apache.qpid.server.security.access.config.ObjectProperties;
+import org.apache.qpid.server.transport.AMQPConnection;
-public interface FirewallRule
+public class ConnectionPrincipalLimitRule extends
ConnectionPrincipalStatisticsRule
{
- boolean matches(InetAddress addressOfClient);
+ public ConnectionPrincipalLimitRule(final int limit)
+ {
+ super(limit);
+ }
- ObjectProperties.Property getPropertyName();
+ protected boolean matches(final AMQPConnection<?> connection)
+ {
+ return connection.getAuthenticatedPrincipalConnectionCount() <=
getLimit();
+ }
- String getPropertyValue();
+ @Override
+ public String toString()
+ {
+ return String.format("ConnectionPrincipalLimitRule{%s=%d}",
+
ObjectProperties.Property.CONNECTION_LIMIT.name().toLowerCase(),
+ getLimit());
+ }
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalStatisticsRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalStatisticsRule.java
new file mode 100644
index 0000000..466eb9f
--- /dev/null
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalStatisticsRule.java
@@ -0,0 +1,70 @@
+/*
+ * 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.config.connection;
+
+import javax.security.auth.Subject;
+
+import org.apache.qpid.server.connection.ConnectionPrincipal;
+import org.apache.qpid.server.security.access.config.DynamicRule;
+import org.apache.qpid.server.transport.AMQPConnection;
+
+public abstract class ConnectionPrincipalStatisticsRule implements DynamicRule
+{
+ private final int _limit;
+
+ ConnectionPrincipalStatisticsRule(final int limit)
+ {
+ _limit = limit;
+ }
+
+ int getLimit()
+ {
+ return _limit;
+ }
+
+ @Override
+ public boolean matches(final Subject subject)
+ {
+ AMQPConnection<?> connection = getConnection(subject);
+ if (connection != null)
+ {
+ return matches(connection);
+ }
+ return false;
+ }
+
+ abstract boolean matches(final AMQPConnection<?> connection);
+
+ private AMQPConnection<?> getConnection(final Subject subject)
+ {
+ if (subject != null)
+ {
+ final ConnectionPrincipal principal =
subject.getPrincipals(ConnectionPrincipal.class)
+ .stream()
+ .findFirst()
+ .orElse(null);
+ if (principal != null)
+ {
+ return principal.getConnection();
+ }
+ }
+ return null;
+ }
+}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
index 72d8d69..d627655 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/FirewallRule.java
@@ -19,14 +19,43 @@
package org.apache.qpid.server.security.access.firewall;
import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Set;
+import javax.security.auth.Subject;
+
+import org.apache.qpid.server.connection.ConnectionPrincipal;
+import org.apache.qpid.server.security.access.config.DynamicRule;
import org.apache.qpid.server.security.access.config.ObjectProperties;
-public interface FirewallRule
+public abstract class FirewallRule implements DynamicRule
{
- boolean matches(InetAddress addressOfClient);
+ @Override
+ public boolean matches(final Subject subject)
+ {
+ final InetAddress addressOfClient = getAddressOfClient(subject);
+ if (addressOfClient == null)
+ {
+ return true;
+ }
+ return matches(addressOfClient);
+ }
+
+ private InetAddress getAddressOfClient(final Subject subject)
+ {
+ final Set<ConnectionPrincipal> principals =
subject.getPrincipals(ConnectionPrincipal.class);
+ if(!principals.isEmpty())
+ {
+ final SocketAddress address =
principals.iterator().next().getConnection().getRemoteSocketAddress();
+ if(address instanceof InetSocketAddress)
+ {
+ return ((InetSocketAddress) address).getAddress();
+ }
+ }
+ return null;
+ }
- ObjectProperties.Property getPropertyName();
+ protected abstract boolean matches(InetAddress addressOfClient);
- String getPropertyValue();
}
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/HostnameFirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/HostnameFirewallRule.java
index dfc0e22..ac95a15 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/HostnameFirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/HostnameFirewallRule.java
@@ -30,9 +30,7 @@ import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
-
-public class HostnameFirewallRule implements FirewallRule
+public class HostnameFirewallRule extends FirewallRule
{
private static final Logger LOGGER =
LoggerFactory.getLogger(HostnameFirewallRule.class);
@@ -58,7 +56,7 @@ public class HostnameFirewallRule implements FirewallRule
}
@Override
- public boolean matches(InetAddress remote)
+ protected boolean matches(InetAddress remote)
{
String hostname = getHostname(remote);
if (hostname == null)
@@ -84,23 +82,11 @@ public class HostnameFirewallRule implements FirewallRule
return false;
}
- @Override
- public ObjectProperties.Property getPropertyName()
- {
- return ObjectProperties.Property.FROM_HOSTNAME;
- }
-
- @Override
- public String getPropertyValue()
- {
- return String.join(",", _hostnames);
- }
-
/**
* @param remote
* the InetAddress to look up
* @return the hostname, null if not found, takes longer than
- * {@value #DNS_LOOKUP} to find or otherwise fails
+ * {@link #DNS_LOOKUP} to find or otherwise fails
*/
private String getHostname(final InetAddress remote) throws
AccessControlFirewallException
{
diff --git
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/NetworkFirewallRule.java
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/NetworkFirewallRule.java
index de7511f..e4d41e8 100644
---
a/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/NetworkFirewallRule.java
+++
b/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/firewall/NetworkFirewallRule.java
@@ -25,17 +25,13 @@ import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.server.security.access.config.ObjectProperties;
-
-public class NetworkFirewallRule implements FirewallRule
+public class NetworkFirewallRule extends FirewallRule
{
private static final Logger LOGGER =
LoggerFactory.getLogger(NetworkFirewallRule.class);
- private final String _originalNetworks;
private List<InetNetwork> _networks;
public NetworkFirewallRule(String... networks)
{
- _originalNetworks = String.join(",", networks);
_networks = new ArrayList<InetNetwork>();
for (int i = 0; i < networks.length; i++)
{
@@ -58,7 +54,7 @@ public class NetworkFirewallRule implements FirewallRule
}
@Override
- public boolean matches(InetAddress ip)
+ protected boolean matches(InetAddress ip)
{
for (InetNetwork network : _networks)
{
@@ -107,16 +103,4 @@ public class NetworkFirewallRule implements FirewallRule
"networks=" + _networks +
']';
}
-
- @Override
- public ObjectProperties.Property getPropertyName()
- {
- return ObjectProperties.Property.FROM_NETWORK;
- }
-
- @Override
- public String getPropertyValue()
- {
- return _originalNetworks;
- }
}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclActionTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclActionTest.java
index d2f4e7c..5e53f0b 100644
---
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclActionTest.java
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclActionTest.java
@@ -89,7 +89,7 @@ public class AclActionTest extends UnitTestBase
private AclRulePredicates createAclRulePredicates()
{
AclRulePredicates predicates = mock(AclRulePredicates.class);
-
when(predicates.getFirewallRule()).thenReturn(mock(FirewallRule.class));
+ when(predicates.getDynamicRule()).thenReturn(mock(FirewallRule.class));
when(predicates.getObjectProperties()).thenReturn(mock(ObjectProperties.class));
return predicates;
}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclRulePredicatesTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclRulePredicatesTest.java
index 0cd767b..b40fe8d 100644
---
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclRulePredicatesTest.java
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/AclRulePredicatesTest.java
@@ -20,16 +20,21 @@ package org.apache.qpid.server.security.access.config;
import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.ATTRIBUTES;
import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.CLASS;
+import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.CONNECTION_LIMIT;
import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.FROM_HOSTNAME;
import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.FROM_NETWORK;
import static
org.apache.qpid.server.security.access.config.ObjectProperties.Property.NAME;
+import static org.hamcrest.CoreMatchers.allOf;
+import static org.hamcrest.collection.IsMapContaining.hasEntry;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.Map;
import java.util.Set;
import org.junit.Before;
@@ -114,4 +119,18 @@ public class AclRulePredicatesTest extends UnitTestBase
_aclRulePredicates.getObjectProperties().getAttributeNames());
}
+
+ @Test
+ public void testGetParsedProperties()
+ {
+ _aclRulePredicates.parse(ATTRIBUTES.name(), "attribute1,attribute2");
+ _aclRulePredicates.parse(FROM_NETWORK.name(), "network1,network2");
+ _aclRulePredicates.parse(CONNECTION_LIMIT.name(), "20");
+
+ final Map<ObjectProperties.Property, String> properties =
_aclRulePredicates.getParsedProperties();
+
+ assertThat(properties, allOf(hasEntry(ATTRIBUTES,
"attribute1,attribute2"),
+ hasEntry(FROM_NETWORK,
"network1,network2"),
+ hasEntry(CONNECTION_LIMIT, "20")));
+ }
}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/ClientActionTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/ClientActionTest.java
index 682aae2..bb5479c 100644
---
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/ClientActionTest.java
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/ClientActionTest.java
@@ -26,68 +26,120 @@ import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.security.Principal;
+import java.util.Collections;
+import java.util.Set;
+import javax.security.auth.Subject;
+
+import org.junit.Before;
import org.junit.Test;
+import org.apache.qpid.server.connection.ConnectionPrincipal;
import org.apache.qpid.server.security.access.firewall.FirewallRule;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.test.utils.UnitTestBase;
public class ClientActionTest extends UnitTestBase
{
- private Action _action = mock(Action.class);
- private AclAction _ruleAction = mock(AclAction.class);
- private InetAddress _addressOfClient = mock(InetAddress.class);
-
- private ClientAction _clientAction = new ClientAction(_action);
+ private Action _action;
+ private AclAction _ruleAction;
+ private ClientAction _clientAction;
+ private Subject _subject;
+ private QueueManagingVirtualHost _addressSpace;
+ private AMQPConnection _connection;
+
+ @Before
+ public void setUp()
+ {
+ final InetSocketAddress address =
InetSocketAddress.createUnresolved("localhost", 5672);
+
+ _addressSpace = mock(QueueManagingVirtualHost.class);
+ _connection = mock(AMQPConnection.class);
+ when(_connection.getAddressSpace()).thenReturn(_addressSpace);
+
+ final ConnectionPrincipal connectionPrincipal =
mock(ConnectionPrincipal.class);
+ when(connectionPrincipal.getConnection()).thenReturn(_connection);
+ when(_connection.getRemoteSocketAddress()).thenReturn(address);
+
when(_connection.getAuthorizedPrincipal()).thenReturn(connectionPrincipal);
+ final Set<? extends Principal> principals =
Collections.singleton(connectionPrincipal);
+ _subject = new Subject(false, principals, Collections.emptySet(),
Collections.emptySet());
+ _ruleAction = mock(AclAction.class);
+ when(_ruleAction.getDynamicRule()).thenReturn(subject -> true);
+ _action = mock(Action.class);
+ _clientAction = new ClientAction(_action);
+ }
@Test
public void testMatches_returnsTrueWhenActionsMatchAndNoFirewallRule()
{
when(_action.matches(any(Action.class))).thenReturn(true);
- when(_ruleAction.getFirewallRule()).thenReturn(null);
+ when(_ruleAction.getDynamicRule()).thenReturn(null);
when(_ruleAction.getAction()).thenReturn(mock(Action.class));
- assertTrue(_clientAction.matches(_ruleAction, _addressOfClient));
+ assertTrue(_clientAction.matches(_ruleAction, _subject));
}
@Test
public void testMatches_returnsFalseWhenActionsDontMatch()
{
FirewallRule firewallRule = mock(FirewallRule.class);
- when(firewallRule.matches(_addressOfClient)).thenReturn(true);
+ when(firewallRule.matches(_subject)).thenReturn(true);
+ when(_ruleAction.getDynamicRule()).thenReturn(firewallRule);
when(_action.matches(any(Action.class))).thenReturn(false);
- when(_ruleAction.getFirewallRule()).thenReturn(firewallRule);
+ when(_ruleAction.getDynamicRule()).thenReturn(firewallRule);
when(_ruleAction.getAction()).thenReturn(mock(Action.class));
- assertFalse(_clientAction.matches(_ruleAction, _addressOfClient));
+ assertFalse(_clientAction.matches(_ruleAction, _subject));
}
@Test
public void testMatches_returnsTrueWhenActionsAndFirewallRuleMatch()
{
FirewallRule firewallRule = mock(FirewallRule.class);
- when(firewallRule.matches(_addressOfClient)).thenReturn(true);
+ when(firewallRule.matches(_subject)).thenReturn(true);
+ when(_ruleAction.getDynamicRule()).thenReturn(firewallRule);
when(_action.matches(any(Action.class))).thenReturn(true);
- when(_ruleAction.getFirewallRule()).thenReturn(firewallRule);
+ when(_ruleAction.getDynamicRule()).thenReturn(firewallRule);
when(_ruleAction.getAction()).thenReturn(mock(Action.class));
- assertTrue(_clientAction.matches(_ruleAction, _addressOfClient));
+ assertTrue(_clientAction.matches(_ruleAction, _subject));
}
@Test
public void testMatches_ignoresFirewallRuleIfClientAddressIsNull()
{
- FirewallRule firewallRule = mock(FirewallRule.class);
+ FirewallRule firewallRule = new FirewallRule()
+ {
+ @Override
+ protected boolean matches(final InetAddress addressOfClient)
+ {
+ return false;
+ }
+ };
when(_action.matches(any(Action.class))).thenReturn(true);
- when(_ruleAction.getFirewallRule()).thenReturn(firewallRule);
+ when(_ruleAction.getDynamicRule()).thenReturn(firewallRule);
when(_ruleAction.getAction()).thenReturn(mock(Action.class));
- assertTrue(_clientAction.matches(_ruleAction, null));
-
- verifyNoInteractions(firewallRule);
+ assertTrue(_clientAction.matches(_ruleAction, _subject));
}
+ @Test
+ public void testMatchesWhenConnectionLimitBreached()
+ {
+ final ObjectProperties properties = new ObjectProperties("foo");
+ final AclRulePredicates predicates = new AclRulePredicates();
+ predicates.parse(ObjectProperties.Property.CONNECTION_LIMIT.name(),
"1");
+ final AclAction ruleAction = new AclAction(LegacyOperation.ACCESS,
ObjectType.VIRTUALHOST, predicates);
+ final ClientAction clientAction = new
ClientAction(LegacyOperation.ACCESS, ObjectType.VIRTUALHOST, properties);
+
+
when(_connection.getAuthenticatedPrincipalConnectionCount()).thenReturn(2);
+ boolean matches = clientAction.matches(ruleAction, _subject);
+ assertFalse(matches);
+ }
}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/DynamicRuleTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/DynamicRuleTest.java
new file mode 100644
index 0000000..9b533bb
--- /dev/null
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/DynamicRuleTest.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.config;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.*;
+
+import javax.security.auth.Subject;
+
+import org.junit.Test;
+
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class DynamicRuleTest extends UnitTestBase
+{
+
+ @Test
+ public void subjectMatchesAllRules()
+ {
+ final Subject subject = new Subject();
+ final DynamicRule rule1 = s -> s.equals(subject);
+ final DynamicRule rule2 = s -> s.equals(subject);
+
+ final DynamicRule combined = rule1.and(rule2);
+
+ assertThat(combined.matches(subject), is(true));
+ }
+
+ @Test
+ public void subjectDoesNotMatchAllRules()
+ {
+ final Subject subject = new Subject();
+ final DynamicRule rule1 = s -> s.equals(subject);
+ final DynamicRule rule2 = s -> !s.equals(subject);
+
+ final DynamicRule combined = rule1.and(rule2);
+
+ assertThat(combined.matches(subject), is(false));
+ }
+}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControlTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControlTest.java
index 02429de..bda1f74 100644
---
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControlTest.java
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/RuleBasedAccessControlTest.java
@@ -272,7 +272,7 @@ public class RuleBasedAccessControlTest extends UnitTestBase
ObjectProperties properties = new
ObjectProperties(testVirtualHost);
accessControl.authorise(LegacyOperation.ACCESS,
ObjectType.VIRTUALHOST, properties);
- verify(mockRuleSet).check(subject, LegacyOperation.ACCESS,
ObjectType.VIRTUALHOST, properties, inetAddress);
+ verify(mockRuleSet).check(subject, LegacyOperation.ACCESS,
ObjectType.VIRTUALHOST, properties);
return null;
}
});
@@ -303,8 +303,7 @@ public class RuleBasedAccessControlTest extends UnitTestBase
subject,
LegacyOperation.ACCESS,
ObjectType.VIRTUALHOST,
- ObjectProperties.EMPTY,
- inetAddress)).thenThrow(new RuntimeException());
+ ObjectProperties.EMPTY)).thenThrow(new
RuntimeException());
RuleBasedAccessControl accessControl = new
RuleBasedAccessControl(mockRuleSet,
BrokerModel.getInstance());
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRuleTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRuleTest.java
new file mode 100644
index 0000000..1619e72
--- /dev/null
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalFrequencyLimitRuleTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.config.connection;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+
+import javax.security.auth.Subject;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.connection.ConnectionPrincipal;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class ConnectionPrincipalFrequencyLimitRuleTest extends UnitTestBase
+{
+
+ private AMQPConnection _connection;
+ private Subject _subject;
+
+ @Before
+ public void setUp()
+ {
+ _connection = mock(AMQPConnection.class);
+
+ final ConnectionPrincipal connectionPrincipal =
mock(ConnectionPrincipal.class);
+ when(connectionPrincipal.getConnection()).thenReturn(_connection);
+
when(_connection.getAuthorizedPrincipal()).thenReturn(connectionPrincipal);
+ _subject = new Subject(false,
+ Collections.singleton(connectionPrincipal),
+ Collections.emptySet(),
+ Collections.emptySet());
+ }
+
+ @Test
+ public void limitBreached()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionFrequency()).thenReturn(2);
+ final ConnectionPrincipalFrequencyLimitRule rule = new
ConnectionPrincipalFrequencyLimitRule(1);
+
+ assertThat(rule.matches(_subject), is(false));
+ }
+
+ @Test
+ public void frequencyEqualsLimit()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionFrequency()).thenReturn(2);
+ final ConnectionPrincipalFrequencyLimitRule rule = new
ConnectionPrincipalFrequencyLimitRule(2);
+
+ assertThat(rule.matches(_subject), is(true));
+ }
+
+ @Test
+ public void frequencyBelowLimit()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionFrequency()).thenReturn(1);
+ final ConnectionPrincipalFrequencyLimitRule rule = new
ConnectionPrincipalFrequencyLimitRule(2);
+
+ assertThat(rule.matches(_subject), is(true));
+ }
+}
diff --git
a/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRuleTest.java
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRuleTest.java
new file mode 100644
index 0000000..4b5598a
--- /dev/null
+++
b/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/config/connection/ConnectionPrincipalLimitRuleTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.config.connection;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+
+import javax.security.auth.Subject;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.connection.ConnectionPrincipal;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public class ConnectionPrincipalLimitRuleTest extends UnitTestBase
+{
+
+ private AMQPConnection _connection;
+ private Subject _subject;
+
+ @Before
+ public void setUp()
+ {
+ _connection = mock(AMQPConnection.class);
+
+ final ConnectionPrincipal connectionPrincipal =
mock(ConnectionPrincipal.class);
+ when(connectionPrincipal.getConnection()).thenReturn(_connection);
+
when(_connection.getAuthorizedPrincipal()).thenReturn(connectionPrincipal);
+ _subject = new Subject(false,
+ Collections.singleton(connectionPrincipal),
+ Collections.emptySet(),
+ Collections.emptySet());
+ }
+
+ @Test
+ public void limitBreached()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionCount()).thenReturn(2);
+ final ConnectionPrincipalLimitRule rule = new
ConnectionPrincipalLimitRule(1);
+
+ assertThat(rule.matches(_subject), is(false));
+ }
+
+ @Test
+ public void frequencyEqualsLimit()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionCount()).thenReturn(2);
+ final ConnectionPrincipalLimitRule rule = new
ConnectionPrincipalLimitRule(2);
+
+ assertThat(rule.matches(_subject), is(true));
+ }
+
+ @Test
+ public void frequencyBelowLimit()
+ {
+
when(_connection.getAuthenticatedPrincipalConnectionCount()).thenReturn(1);
+ final ConnectionPrincipalLimitRule rule = new
ConnectionPrincipalLimitRule(2);
+
+ assertThat(rule.matches(_subject), is(true));
+ }
+}
diff --git
a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
index d5bd980..f7e0259 100644
---
a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
+++
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
@@ -26,6 +26,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
import java.util.List;
import javax.security.auth.Subject;
@@ -40,6 +42,7 @@ import
org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.configuration.updater.TaskExecutorImpl;
import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.AuthenticationProvider;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
import org.apache.qpid.server.model.BrokerTestHelper;
@@ -54,6 +57,8 @@ import
org.apache.qpid.server.protocol.v0_10.transport.ExecutionErrorCode;
import org.apache.qpid.server.protocol.v0_10.transport.ExecutionException;
import org.apache.qpid.server.protocol.v0_10.transport.MessageTransfer;
import org.apache.qpid.server.protocol.v0_10.transport.Method;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.security.auth.UsernamePrincipal;
import org.apache.qpid.test.utils.UnitTestBase;
public class ServerSessionTest extends UnitTestBase
@@ -120,9 +125,13 @@ public class ServerSessionTest extends UnitTestBase
when(modelConnection.getModel()).thenReturn(BrokerModel.getInstance());
when(modelConnection.getPort()).thenReturn(port);
- Subject subject = new Subject();
+ final AuthenticatedPrincipal principal =
+ new AuthenticatedPrincipal(new
UsernamePrincipal(getTestName(), mock(AuthenticationProvider.class)));
+ final Subject subject =
+ new Subject(false, Collections.singleton(principal),
Collections.emptySet(), Collections.emptySet());
when(modelConnection.getSubject()).thenReturn(subject);
when(modelConnection.getMaxMessageSize()).thenReturn(1024l);
+ when(modelConnection.getCreatedTime()).thenReturn(new Date());
ServerConnection connection = new ServerConnection(1, broker, port,
Transport.TCP, modelConnection);
connection.setVirtualHost(_virtualHost);
diff --git
a/doc/java-broker/src/docbkx/security/Java-Broker-Security-AccessControlProviders.xml
b/doc/java-broker/src/docbkx/security/Java-Broker-Security-AccessControlProviders.xml
index 0fecf48..37e5bfa 100644
---
a/doc/java-broker/src/docbkx/security/Java-Broker-Security-AccessControlProviders.xml
+++
b/doc/java-broker/src/docbkx/security/Java-Broker-Security-AccessControlProviders.xml
@@ -283,7 +283,7 @@
<entry> <command>VIRTUALHOST</command> </entry>
<entry> <para>A virtualhost</para> </entry>
<entry><para>ALL, CREATE, UPDATE, DELETE, ACCESS, ACCESS_LOGS,
INVOKE</para> </entry>
- <entry><para>name</para> </entry>
+ <entry><para>name, connection_limit,
connection_frequency_limit</para> </entry>
<entry><para>No</para> </entry>
</row>
<row>
@@ -415,6 +415,36 @@
</entry>
</row>
<row>
+ <entry><command>connection_limit</command></entry>
+ <entry>
+ <para>
+ A maximum number of connections the users belonging to the given
identity can establish to the Virtual Host
+ </para>
+ <para>
+ Intended for use in ACCESS VIRTUALHOST rules to restrict the
number of connections which can be made
+ by the messaging user.
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry><command>connection_frequency_limit</command></entry>
+ <entry>
+ <para>
+ A maximum number of connections the user belonging to the given
identity can establish to the Virtual Host
+ within pre-defined period of time, which is 1 minute by default.
+ </para>
+ <para>
+ Intended for use in ACCESS VIRTUALHOST rules to restrict the
frequency of connections which can be made
+ by the messaging user.
+ </para>
+ <para>If required, the frequency period can be changed using
context variable
+
<emphasis>qpid.virtualhost.connectionFrequencyPeriodInMillis</emphasis>. As
name suggests, its value needs
+ to be specified in milliseconds. Setting it to zero or negative
value turns off the
+ <command>connection_frequency</command> evaluation.
+ </para>
+ </entry>
+ </row>
+ <row>
<entry><command>virtualhost_name</command></entry>
<entry>
<para>
@@ -478,6 +508,28 @@
</example>
</section>
+ <section role="h4"
xml:id="Java-Broker-Security-AccessControlProviders-ResourceRestrictions">
+ <title>
+ Resource restrictions
+ </title>
+ <para>
+ The number of Virtual Host connections and frequency of connections
can be restricted with
+ <emphasis>ACCESS</emphasis> rule.
+ </para>
+ <example>
+ <title>Restricting user connections to virtual host</title>
+ <programlisting><![CDATA[
+ ACL ALLOW-LOG guest ACCESS VIRTUALHOST connection_limit=1
+ ACL ALLOW-LOG messaging-users ACCESS VIRTUALHOST
connection_frequency_limit=100
+ ...]]>
+ </programlisting>
+ </example>
+ <para>In the example above the maximum number of connections allowed to
establish to the Virtual Host by user
+ <emphasis>guest</emphasis> is set to <emphasis>1</emphasis>. The
maximum connection frequency for every user
+ belonging to the user group <emphasis>messaging-users</emphasis> is
set to <emphasis>100</emphasis> (per minute,
+ by default).
+ </para>
+ </section>
<section role="h4"
xml:id="Java-Broker-Security-AccessControlProviders-WorkedExample2">
<title>
Worked example 2 - Simple Messaging
diff --git
a/systests/qpid-systests-jms_1.1/src/test/java/org/apache/qpid/systests/jms_1_1/extensions/acl/MessagingACLTest.java
b/systests/qpid-systests-jms_1.1/src/test/java/org/apache/qpid/systests/jms_1_1/extensions/acl/MessagingACLTest.java
index 8a9fecd..cb591e8 100644
---
a/systests/qpid-systests-jms_1.1/src/test/java/org/apache/qpid/systests/jms_1_1/extensions/acl/MessagingACLTest.java
+++
b/systests/qpid-systests-jms_1.1/src/test/java/org/apache/qpid/systests/jms_1_1/extensions/acl/MessagingACLTest.java
@@ -51,6 +51,7 @@ import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
+import javax.naming.NamingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
@@ -70,9 +71,6 @@ import
org.apache.qpid.server.security.access.config.ObjectProperties;
import org.apache.qpid.server.security.access.config.ObjectType;
import org.apache.qpid.server.security.access.config.Rule;
import org.apache.qpid.server.security.access.config.RuleSet;
-import org.apache.qpid.server.security.access.firewall.FirewallRule;
-import org.apache.qpid.server.security.access.firewall.HostnameFirewallRule;
-import org.apache.qpid.server.security.access.firewall.NetworkFirewallRule;
import org.apache.qpid.server.security.access.plugins.AclRule;
import
org.apache.qpid.server.security.access.plugins.RuleBasedVirtualHostAccessControlProvider;
import org.apache.qpid.server.security.access.plugins.RuleOutcome;
@@ -184,6 +182,125 @@ public class MessagingACLTest extends JmsTestBase
}
@Test
+ public void testAuthorizationWithConnectionLimit() throws Exception
+ {
+ final int connectionLimit = 2;
+ configureACL(String.format("ACL ALLOW-LOG %s ACCESS VIRTUALHOST
connection_limit='%d'",
+ USER2,
+ connectionLimit));
+
+ final List<Connection> establishedConnections = new ArrayList<>();
+ try
+ {
+ establishConnections(connectionLimit, establishedConnections);
+
+ verifyConnectionEstablishmentFails(connectionLimit);
+
+ establishedConnections.remove(0).close();
+
getConnectionBuilder().setUsername(USER2).setPassword(USER2_PASSWORD).build().close();
+ }
+ finally
+ {
+ closeConnections(establishedConnections);
+ }
+ }
+
+ @Test
+ public void testAuthorizationWithConnectionFrequencyLimit() throws
Exception
+ {
+ final int connectionFrequencyLimit = 1;
+ configureACL(String.format("ACL ALLOW-LOG %s ACCESS VIRTUALHOST
connection_frequency_limit='%d'",
+ USER2,
+ connectionFrequencyLimit));
+
+ final List<Connection> establishedConnections = new ArrayList<>();
+ try
+ {
+ establishConnections(connectionFrequencyLimit,
establishedConnections);
+
+ verifyConnectionEstablishmentFails(connectionFrequencyLimit);
+
+ establishedConnections.remove(0).close();
+
+ verifyConnectionEstablishmentFails(connectionFrequencyLimit);
+ }
+ finally
+ {
+ closeConnections(establishedConnections);
+ }
+ }
+
+ @Test
+ public void testAuthorizationWithConnectionLimitAndFrequencyLimit() throws
Exception
+ {
+ final int connectionFrequencyLimit = 2;
+ final int connectionLimit = 3;
+ configureACL(String.format("ACL ALLOW-LOG %s ACCESS VIRTUALHOST
connection_limit='%d' connection_frequency_limit='%d'",
+ USER2,
+ connectionLimit,
+ connectionFrequencyLimit));
+
+ final List<Connection> establishedConnections = new ArrayList<>();
+ try
+ {
+ establishConnections(connectionFrequencyLimit,
establishedConnections);
+
+ verifyConnectionEstablishmentFails(connectionFrequencyLimit);
+
+ establishedConnections.remove(0).close();
+
+ verifyConnectionEstablishmentFails(connectionFrequencyLimit);
+ }
+ finally
+ {
+ closeConnections(establishedConnections);
+ }
+ }
+
+ private void establishConnections(final int connectionNumber, final
List<Connection> establishedConnections)
+ throws NamingException, JMSException
+ {
+ for (int i = 0; i < connectionNumber; i++)
+ {
+
establishedConnections.add(getConnectionBuilder().setUsername(USER2)
+
.setPassword(USER2_PASSWORD)
+
.setClientId(getTestName() + i)
+ .build());
+ }
+ }
+
+ private void closeConnections(final List<Connection>
establishedConnections) throws JMSException
+ {
+ for (Connection c : establishedConnections)
+ {
+ c.close();
+ }
+ }
+
+ private void verifyConnectionEstablishmentFails(final int frequencyLimit)
throws NamingException
+ {
+ try
+ {
+ final Connection connection =
getConnectionBuilder().setUsername(USER2)
+
.setPassword(USER2_PASSWORD)
+
.setClientId(getTestName() + frequencyLimit)
+ .build();
+ try
+ {
+ fail("Connection creation should fail due to exceeding limit");
+ }
+ finally
+ {
+ connection.close();
+ }
+ }
+ catch (JMSException e)
+ {
+ //pass
+ }
+ }
+
+ @Test
public void testConsumeFromTempQueueSuccess() throws Exception
{
configureACL(String.format("ACL ALLOW-LOG %s ACCESS VIRTUALHOST",
USER1),
@@ -865,21 +982,7 @@ public class MessagingACLTest extends JmsTestBase
@Override
public Map<ObjectProperties.Property, String>
getAttributes()
{
- Map<ObjectProperties.Property, String> attributes =
new HashMap<>(rule.getAction().getProperties().asPropertyMap());
- FirewallRule firewallRule =
rule.getAclAction().getFirewallRule();
- if (firewallRule != null)
- {
- if (firewallRule instanceof HostnameFirewallRule)
- {
-
attributes.put(ObjectProperties.Property.FROM_HOSTNAME, "127.0.0.1");
- }
- else if (firewallRule instanceof
NetworkFirewallRule)
- {
- // tests use only 127.0.0.1 at the moment
-
attributes.put(ObjectProperties.Property.FROM_NETWORK, "127.0.0.1");
- }
- }
- return attributes;
+ return rule.getAttributes();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
