diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java index 43f8d7f9..966cb878 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java @@ -29,11 +29,13 @@ import org.eclipse.moquette.server.ServerChannel; public class NettyChannel implements ServerChannel { private ChannelHandlerContext m_channel; - + + public static final String ATTR_USERNAME = "username"; public static final AttributeKey ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE); public static final AttributeKey ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION); public static final AttributeKey ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID); + public static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); NettyChannel(ChannelHandlerContext ctx) { m_channel = ctx; @@ -48,7 +50,7 @@ public class NettyChannel implements ServerChannel { Attribute attr = m_channel.attr(key); attr.set(value); } - + public void setIdleTime(int idleTime) { if (m_channel.pipeline().names().contains("idleStateHandler")) { m_channel.pipeline().remove("idleStateHandler"); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 4217921c..fe730e00 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -215,6 +215,7 @@ class ProtocolProcessor implements EventHandler { failedCredetials(session); return; } + session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername()); } else if (!this.allowAnonymous) { failedCredetials(session); return; @@ -297,7 +298,8 @@ class ProtocolProcessor implements EventHandler { final ByteBuffer message = msg.getPayload(); boolean retain = msg.isRetainFlag(); //check if the topic can be wrote - if (m_authorizator.canWrite(topic)) { + String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); + if (m_authorizator.canWrite(topic, user)) { processPublish(clientID, topic, qos, message, retain, msg.getMessageID()); } else { LOG.debug("topic {} doesn't have write credentials", topic); @@ -373,11 +375,7 @@ class ProtocolProcessor implements EventHandler { if (qos.ordinal() > sub.getRequestedQos().ordinal()) { qos = sub.getRequestedQos(); } - if (!m_authorizator.canRead(topic)) { - LOG.debug("topic {} doesn't have read credentials", topic); - continue; - } - + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive()); ByteBuffer message = origMessage.duplicate(); @@ -437,9 +435,16 @@ class ProtocolProcessor implements EventHandler { if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } - LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession()); -// m_clientIDs.get(clientId).getSession().write(pubMessage); - disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage)); + ServerChannel session = m_clientIDs.get(clientId).getSession(); + LOG.debug("Session for clientId {} is {}", clientId, session); + + String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); + if (!m_authorizator.canRead(topic, user)) { + LOG.debug("topic {} doesn't have read credentials", topic); + return; + } + + disruptorPublish(new OutputMessagingEvent(session, pubMessage)); } private void sendPubRec(String clientID, int messageID) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollector.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollector.java index 90d0ce1f..18c6d2d9 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollector.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollector.java @@ -20,9 +20,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.text.ParseException; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; +import java.util.*; import static org.eclipse.moquette.spi.impl.security.Authorization.Permission.READ; import static org.eclipse.moquette.spi.impl.security.Authorization.Permission.READWRITE; @@ -40,55 +38,94 @@ class AuthorizationsCollector implements IAuthorizator { private static final Logger LOG = LoggerFactory.getLogger(AuthorizationsCollector.class); - private List m_authorizations = new ArrayList(); + private List m_globalAuthorizations = new ArrayList(); + private List m_patternAuthorizations = new ArrayList(); + private Map> m_userAuthorizations = new HashMap(); + private boolean m_parsingUsersSpecificSection = false; + private String m_currentUser = ""; static final AuthorizationsCollector emptyImmutableCollector() { AuthorizationsCollector coll = new AuthorizationsCollector(); - coll.m_authorizations = Collections.emptyList(); + coll.m_globalAuthorizations = Collections.emptyList(); + coll.m_patternAuthorizations = Collections.emptyList(); + coll.m_userAuthorizations = Collections.emptyMap(); return coll; } void parse(String line) throws ParseException { - m_authorizations.add(parseAuthLine(line)); + Authorization acl = parseAuthLine(line); + if (acl == null) { + //skip it's a user + return; + } + if (m_parsingUsersSpecificSection) { + //TODO in java 8 switch to m_userAuthorizations.putIfAbsent(m_currentUser, new ArrayList()); + if (!m_userAuthorizations.containsKey(m_currentUser)) { + m_userAuthorizations.put(m_currentUser, new ArrayList()); + } + List userAuths = m_userAuthorizations.get(m_currentUser); + userAuths.add(acl); + } else { + m_globalAuthorizations.add(acl); + } } - protected static Authorization parseAuthLine(String line) throws ParseException { + protected Authorization parseAuthLine(String line) throws ParseException { String[] tokens = line.split("\\s+"); - String keyword = tokens[0]; - if ("topic".equalsIgnoreCase(keyword)) { - if (tokens.length > 2) { - //if the tokenized lines has 3 token the second must be the permission - try { - Authorization.Permission permission = Authorization.Permission.valueOf(tokens[1].toUpperCase()); - //bring topic with all original spacing - String topic = line.substring(line.indexOf(tokens[2])); + String keyword = tokens[0].toLowerCase(); + switch (keyword) { + case "topic": + if (tokens.length > 2) { + //if the tokenized lines has 3 token the second must be the permission + try { + Authorization.Permission permission = Authorization.Permission.valueOf(tokens[1].toUpperCase()); + //bring topic with all original spacing + String topic = line.substring(line.indexOf(tokens[2])); - return new Authorization(topic, permission); - } catch (IllegalArgumentException iaex) { - throw new ParseException("invalid permission token", 1); + return new Authorization(topic, permission); + } catch (IllegalArgumentException iaex) { + throw new ParseException("invalid permission token", 1); + } + } + String topic = tokens[1]; + return new Authorization(topic); + case "user": + m_parsingUsersSpecificSection = true; + m_currentUser = tokens[1]; + return null; + case "pattern": + //TODO implement the part for patter matching + return null; + default: + throw new ParseException(String.format("invalid line definition found %s", line), 1); + } + } + + @Override + public boolean canWrite(String topic, String user) { + return canDoOperation(topic, WRITE, user); + } + + @Override + public boolean canRead(String topic, String user) { + return canDoOperation(topic, READ, user); + } + + private boolean canDoOperation(String topic, Authorization.Permission permission, String username) { + if (username == null || username.isEmpty()) { + for (Authorization auth : m_globalAuthorizations) { + if (auth.permission == permission || auth.permission == READWRITE) { + if (SubscriptionsStore.matchTopics(topic, auth.topic)) { + return true; + } } } - String topic = tokens[1]; - return new Authorization(topic); - } - return null; - } - - @Override - public boolean canWrite(String topic) { - return canDoOperation(topic, WRITE); - } - - @Override - public boolean canRead(String topic) { - return canDoOperation(topic, READ); - } - - private boolean canDoOperation(String topic, Authorization.Permission permission) { - for (Authorization auth : m_authorizations) { - if (auth.permission == permission || auth.permission == READWRITE) { - if (SubscriptionsStore.matchTopics(topic, auth.topic)) { - return true; + } else if (m_userAuthorizations.containsKey(username)) { + for (Authorization auth : m_userAuthorizations.get(username)) { + if (auth.permission == permission || auth.permission == READWRITE) { + if (SubscriptionsStore.matchTopics(topic, auth.topic)) { + return true; + } } } } @@ -96,6 +133,6 @@ class AuthorizationsCollector implements IAuthorizator { } public boolean isEmpty() { - return m_authorizations.isEmpty(); + return m_globalAuthorizations.isEmpty(); } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/DenyAllAuthorizator.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/DenyAllAuthorizator.java index ff394e3e..0d96284a 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/DenyAllAuthorizator.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/DenyAllAuthorizator.java @@ -20,12 +20,12 @@ package org.eclipse.moquette.spi.impl.security; */ public class DenyAllAuthorizator implements IAuthorizator { @Override - public boolean canWrite(String topic) { + public boolean canWrite(String topic, String user) { return true; } @Override - public boolean canRead(String topic) { + public boolean canRead(String topic, String user) { return true; } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java index 39d64dc1..39fa5b1e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java @@ -29,7 +29,7 @@ public interface IAuthorizator { /** * Ask the implementation of the authorizator if the topic can be used in a publish. * */ - boolean canWrite(String topic); + boolean canWrite(String topic, String user); - boolean canRead(String topic); + boolean canRead(String topic, String user); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java index a22a8aca..3f9987aa 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java @@ -20,12 +20,12 @@ package org.eclipse.moquette.spi.impl.security; */ public class PermitAllAuthorizator implements IAuthorizator { @Override - public boolean canWrite(String topic) { + public boolean canWrite(String topic, String user) { return true; } @Override - public boolean canRead(String topic) { + public boolean canRead(String topic, String user) { return true; } } diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java index dc708fe1..9ac3cf9b 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -29,7 +29,6 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 9562e4dd..7326bff2 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -99,7 +99,7 @@ public class ProtocolProcessorTest { throw new AssertionError("Wrong return code"); } } - } + } /** * This a synchronous channel that avoid output ring buffer from Processor @@ -141,7 +141,7 @@ public class ProtocolProcessorTest { throw new AssertionError("Wrong return code"); } } - } + } @Before public void setUp() throws InterruptedException { diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/security/ACLFileParserTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/security/ACLFileParserTest.java index d82bc374..b43635f9 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/security/ACLFileParserTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/security/ACLFileParserTest.java @@ -20,8 +20,6 @@ import org.junit.Test; import java.io.Reader; import java.io.StringReader; import java.text.ParseException; -import java.util.List; -import static org.eclipse.moquette.spi.impl.security.Authorization.*; import static org.junit.Assert.*; @@ -60,8 +58,8 @@ public class ACLFileParserTest { AuthorizationsCollector authorizations = ACLFileParser.parse(conf); //Verify - assertTrue(authorizations.canRead("/weather/italy/anemometer")); - assertTrue(authorizations.canWrite("/weather/italy/anemometer")); + assertTrue(authorizations.canRead("/weather/italy/anemometer", "")); + assertTrue(authorizations.canWrite("/weather/italy/anemometer", "")); } } \ No newline at end of file diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollectorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollectorTest.java index 5020e0d1..04f7efb2 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollectorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollectorTest.java @@ -20,7 +20,6 @@ import org.junit.Test; import java.text.ParseException; -import static org.eclipse.moquette.spi.impl.security.AuthorizationsCollector.*; import static org.junit.Assert.*; /** @@ -41,7 +40,7 @@ public class AuthorizationsCollectorTest { @Test public void testParseAuthLineValid() throws ParseException { - Authorization authorization = parseAuthLine("topic /weather/italy/anemometer"); + Authorization authorization = authorizator.parseAuthLine("topic /weather/italy/anemometer"); //Verify assertEquals(RW_ANEMOMETER, authorization); @@ -49,7 +48,7 @@ public class AuthorizationsCollectorTest { @Test public void testParseAuthLineValid_read() throws ParseException { - Authorization authorization = parseAuthLine("topic read /weather/italy/anemometer"); + Authorization authorization = authorizator.parseAuthLine("topic read /weather/italy/anemometer"); //Verify assertEquals(R_ANEMOMETER, authorization); @@ -57,7 +56,7 @@ public class AuthorizationsCollectorTest { @Test public void testParseAuthLineValid_write() throws ParseException { - Authorization authorization = parseAuthLine("topic write /weather/italy/anemometer"); + Authorization authorization = authorizator.parseAuthLine("topic write /weather/italy/anemometer"); //Verify assertEquals(W_ANEMOMETER, authorization); @@ -65,7 +64,7 @@ public class AuthorizationsCollectorTest { @Test public void testParseAuthLineValid_readwrite() throws ParseException { - Authorization authorization = parseAuthLine("topic readwrite /weather/italy/anemometer"); + Authorization authorization = authorizator.parseAuthLine("topic readwrite /weather/italy/anemometer"); //Verify assertEquals(RW_ANEMOMETER, authorization); @@ -75,7 +74,7 @@ public class AuthorizationsCollectorTest { @Test public void testParseAuthLineValid_topic_with_space() throws ParseException { Authorization expected = new Authorization("/weather/eastern italy/anemometer"); - Authorization authorization = parseAuthLine("topic readwrite /weather/eastern italy/anemometer"); + Authorization authorization = authorizator.parseAuthLine("topic readwrite /weather/eastern italy/anemometer"); //Verify assertEquals(expected, authorization); @@ -83,7 +82,7 @@ public class AuthorizationsCollectorTest { @Test(expected = ParseException.class) public void testParseAuthLineValid_invalid() throws ParseException { - parseAuthLine("topic faker /weather/italy/anemometer"); + authorizator.parseAuthLine("topic faker /weather/italy/anemometer"); } @Test @@ -91,7 +90,7 @@ public class AuthorizationsCollectorTest { authorizator.parse("topic write /sensors"); //verify - assertTrue(authorizator.canWrite("/sensors")); + assertTrue(authorizator.canWrite("/sensors", "")); } @Test @@ -99,7 +98,7 @@ public class AuthorizationsCollectorTest { authorizator.parse("topic read /sensors"); //verify - assertTrue(authorizator.canRead("/sensors")); + assertTrue(authorizator.canRead("/sensors", "")); } @Test @@ -108,8 +107,8 @@ public class AuthorizationsCollectorTest { authorizator.parse("topic read /sensors/anemometer"); //verify - assertTrue(authorizator.canWrite("/sensors")); - assertFalse(authorizator.canRead("/sensors")); + assertTrue(authorizator.canWrite("/sensors", "")); + assertFalse(authorizator.canRead("/sensors", "")); } @Test @@ -117,7 +116,7 @@ public class AuthorizationsCollectorTest { authorizator.parse("topic write /sensors/#"); //verify - assertTrue(authorizator.canWrite("/sensors/anemometer/wind")); + assertTrue(authorizator.canWrite("/sensors/anemometer/wind", "")); } @Test @@ -125,6 +124,16 @@ public class AuthorizationsCollectorTest { authorizator.parse("topic write /sensors/+"); //verify - assertTrue(authorizator.canWrite("/sensors/anemometer")); + assertTrue(authorizator.canWrite("/sensors/anemometer", "")); + } + + @Test + public void testCanWriteUserTopic() throws ParseException { + authorizator.parse("user john"); + authorizator.parse("topic write /sensors"); + + //verify + assertTrue(authorizator.canWrite("/sensors", "john")); + assertFalse(authorizator.canWrite("/sensors", "jack")); } } \ No newline at end of file