anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added plumbing code to use the concept of Authorizator into publish semantic

Browse Source
This commit is contained in:
andrea
2015-05-17 16:24:37 +02:00
parent 6d3fbc4591
commit f3ed351c61
16 changed files with 186 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/ConfigurationParser.java
+1 -1
View File
@@ -59,7 +59,7 @@ class ConfigurationParser {
m_properties.put(WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(WEBSOCKET_PORT));
m_properties.put(PASSWORD_FILE_PROPERTY_NAME, "");
m_properties.put(PERSISTENT_STORE_PROPERTY_NAME, DEFAULT_PERSISTENT_PATH);
m_properties.put(ALLOW_ANONYMOUS, true);
m_properties.put(ALLOW_ANONYMOUS_PROPERTY_NAME, true);
}
/**
broker/src/main/java/org/eclipse/moquette/spi/impl/AcceptAllAuthenticator.java
-12
View File
@@ -1,12 +0,0 @@
package org.eclipse.moquette.spi.impl;
import org.eclipse.moquette.server.IAuthenticator;
/**
* Created by andrea on 8/23/14.
*/
class AcceptAllAuthenticator implements IAuthenticator {
public boolean checkValid(String username, String password) {
return true;
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+13 -6
View File
@@ -28,6 +28,7 @@ import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.*;
import org.eclipse.moquette.spi.impl.security.IAuthorizator;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1;
@@ -47,7 +48,7 @@ import org.eclipse.moquette.proto.messages.SubscribeMessage;
import org.eclipse.moquette.proto.messages.UnsubAckMessage;
import org.eclipse.moquette.proto.messages.UnsubscribeMessage;
import org.eclipse.moquette.server.ConnectionDescriptor;
import org.eclipse.moquette.server.IAuthenticator;
import org.eclipse.moquette.spi.impl.security.IAuthenticator;
import org.eclipse.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -98,6 +99,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<>();
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
private IAuthorizator m_authorizator;
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
@@ -117,14 +119,16 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
* @param sessionsStore the clients sessions store, used to persist subscriptions.
* @param authenticator the authenticator used in connect messages.
* @param allowAnonymous true connection to clients without credentials.
* @param authorizator used to apply ACL policies to publishes and subscriptions.
*/
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous) {
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator) {
//m_clientIDs = clientIDs;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
m_authorizator = authorizator;
LOG.debug("subscription tree on init {}", subscriptions.dumpTree());
m_authenticator = authenticator;
m_messagesStore = storageService;
@@ -292,7 +296,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
final AbstractMessage.QOSType qos = msg.getQos();
final ByteBuffer message = msg.getPayload();
boolean retain = msg.isRetainFlag();
processPublish(clientID, topic, qos, message, retain, msg.getMessageID());
//check if the topic can be wrote
if (m_authorizator.canWrite(topic)) {
processPublish(clientID, topic, qos, message, retain, msg.getMessageID());
}
}
private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) {
broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java
+17 -4
View File
@@ -20,7 +20,7 @@ import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.HdrHistogram.Histogram;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.server.IAuthenticator;
import org.eclipse.moquette.spi.impl.security.*;
import org.eclipse.moquette.server.ServerChannel;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.IMessaging;
@@ -34,6 +34,10 @@ import org.eclipse.moquette.spi.persistence.MapDBPersistentStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.text.ParseException;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -42,7 +46,8 @@ import java.util.concurrent.TimeUnit;
import static org.eclipse.moquette.commons.Constants.PASSWORD_FILE_PROPERTY_NAME;
import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME;
import static org.eclipse.moquette.commons.Constants.ALLOW_ANONYMOUS;
import static org.eclipse.moquette.commons.Constants.ALLOW_ANONYMOUS_PROPERTY_NAME;
import static org.eclipse.moquette.commons.Constants.ACL_FILE_PROPERTY_NAME;
/**
*
@@ -198,8 +203,16 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
authenticator = new FileAuthenticator(configPath, passwdPath);
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS, "true"));
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous);
String aclFilePath = props.getProperty(ACL_FILE_PROPERTY_NAME, "");
IAuthorizator authorizator;
if (aclFilePath != null && !aclFilePath.isEmpty()) {
authorizator = new ACLAuthorizator(aclFilePath);
} else {
authorizator = new PermitAllAuthorizator();
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/ACLAuthorizator.java
+48
View File
@@ -0,0 +1,48 @@
/*
* Copyright (c) 2012-2014 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi.impl.security;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.text.ParseException;
import java.util.Collections;
import java.util.List;
/**
* @author andrea
*/
public class ACLAuthorizator implements IAuthorizator {
private static final Logger LOG = LoggerFactory.getLogger(ACLAuthorizator.class);
List<Authorization> authorisations = Collections.emptyList();
public ACLAuthorizator(String aclFilePath) {
File aclFile = new File(aclFilePath);
try {
authorisations = ACLFileParser.parse(aclFile);
} catch (ParseException pex) {
LOG.warn(String.format("Format error in parsing acl file %s", aclFile), pex);
}
}
@Override
public boolean canWrite(String topic) {
return false;
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/ACLFileParser.java
+22 -3
View File
@@ -18,9 +18,7 @@ package org.eclipse.moquette.spi.impl.security;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.io.*;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
@@ -34,6 +32,27 @@ import static org.eclipse.moquette.spi.impl.security.Authorization.Permission;
public class ACLFileParser {
private static final Logger LOG = LoggerFactory.getLogger(ACLFileParser.class);
/**
* Parse the configuration from file.
*/
public static List<Authorization> parse(File file) throws ParseException {
if (file == null) {
LOG.warn("parsing NULL file, so fallback on default configuration!");
return Collections.emptyList();
}
if (!file.exists()) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()));
return Collections.emptyList();
}
try {
FileReader reader = new FileReader(file);
return parse(reader);
} catch (FileNotFoundException fex) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()), fex);
return Collections.emptyList();
}
}
/**
* Parse the ACL configuration file
*
broker/src/main/java/org/eclipse/moquette/spi/impl/security/AcceptAllAuthenticator.java
+10
View File
@@ -0,0 +1,10 @@
package org.eclipse.moquette.spi.impl.security;
/**
* Created by andrea on 8/23/14.
*/
public class AcceptAllAuthenticator implements IAuthenticator {
public boolean checkValid(String username, String password) {
return true;
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/Authorization.java
+1 -1
View File
@@ -20,7 +20,7 @@ package org.eclipse.moquette.spi.impl.security;
*
* @author andrea
*/
class Authorization {
public class Authorization {
private String topic;
private Permission permission;
broker/src/main/java/org/eclipse/moquette/spi/impl/FileAuthenticator.java → broker/src/main/java/org/eclipse/moquette/spi/impl/security/FileAuthenticator.java
+2 -3
View File
@@ -13,10 +13,9 @@
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi.impl;
package org.eclipse.moquette.spi.impl.security;
import org.apache.commons.codec.binary.Hex;
import org.eclipse.moquette.server.IAuthenticator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,7 +45,7 @@ public class FileAuthenticator implements IAuthenticator {
private Map<String, String> m_identities = new HashMap<>();
private MessageDigest m_digest;
FileAuthenticator(String parent, String filePath) {
public FileAuthenticator(String parent, String filePath) {
File file = new File(parent, filePath);
LOG.info("Loading password file: " + file);
if (file.isDirectory()) {
broker/src/main/java/org/eclipse/moquette/server/IAuthenticator.java → broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthenticator.java
+1 -1
View File
@@ -13,7 +13,7 @@
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.server;
package org.eclipse.moquette.spi.impl.security;
/**
* username and password checker
broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java
+29
View File
@@ -0,0 +1,29 @@
/*
* Copyright (c) 2012-2014 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi.impl.security;
/**
* ACL checker
*
* @author andrea
*/
public interface IAuthorizator {
/**
* Ask the implementation of the authorizator if the topic can be used in a publish.
* */
boolean canWrite(String topic);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java
+26
View File
@@ -0,0 +1,26 @@
/*
* Copyright (c) 2012-2014 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi.impl.security;
/**
* @author andrea
*/
public class PermitAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(String topic) {
return true;
}
}
broker/src/test/java/org/eclipse/moquette/spi/impl/MockAuthenticator.java
+1 -1
View File
@@ -16,7 +16,7 @@
package org.eclipse.moquette.spi.impl;
import java.util.Map;
import org.eclipse.moquette.server.IAuthenticator;
import org.eclipse.moquette.spi.impl.security.IAuthenticator;
/**
* Test utility to implements authenticator instance.
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java
+12 -12
View File
@@ -27,6 +27,7 @@ import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.LostConnectionEvent;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1_1;
@@ -38,7 +39,6 @@ import org.eclipse.moquette.proto.messages.DisconnectMessage;
import org.eclipse.moquette.proto.messages.PublishMessage;
import org.eclipse.moquette.proto.messages.SubAckMessage;
import org.eclipse.moquette.proto.messages.SubscribeMessage;
import org.eclipse.moquette.server.Constants;
import org.eclipse.moquette.server.ServerChannel;
import static org.junit.Assert.*;
import org.junit.Before;
@@ -164,7 +164,7 @@ public class ProtocolProcessorTest {
subscriptions = new SubscriptionsStore();
subscriptions.init(new MemoryStorageService());
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true);
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator());
}
@Test
@@ -255,7 +255,7 @@ public class ProtocolProcessorTest {
@Test
public void prohibitAnonymousClient() {
connMsg.setClientID("123");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false);
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator());
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -269,7 +269,7 @@ public class ProtocolProcessorTest {
connMsg.setClientID("123");
connMsg.setUserFlag(true);
connMsg.setUsername(TEST_USER + "_fake");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false);
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator());
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -281,7 +281,7 @@ public class ProtocolProcessorTest {
@Test
public void acceptAnonymousClient() {
connMsg.setClientID("123");
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true);
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator());
//Exercise
m_processor.processConnect(m_session, connMsg);
@@ -336,7 +336,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null, true);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
connectMessage.setClientID(FAKE_CLIENT_ID);
@@ -380,7 +380,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null, true);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
ConnectMessage connectMessage = new ConnectMessage();
@@ -498,7 +498,7 @@ public class ProtocolProcessorTest {
subs.init(new MemoryStorageService());
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService, m_sessionStore, null, true);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_PUBLISHER_ID);
connectMessage.setProcotolVersion((byte)3);
@@ -537,7 +537,7 @@ public class ProtocolProcessorTest {
ByteBuffer.wrap("Hello".getBytes()), true, FAKE_PUBLISHER_ID, 120);
m_storageService.storePublishForFuture(retainedMessage);
m_processor.init(subs, m_storageService, m_sessionStore, null, true);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_PUBLISHER_ID);
connectMessage.setProcotolVersion((byte)3);
@@ -556,7 +556,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true);
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
@@ -581,7 +581,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true);
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
@@ -649,7 +649,7 @@ public class ProtocolProcessorTest {
/*SubscriptionsStore subs = new SubscriptionsStore();
subs.init(new MemoryStorageService());*/
SubscriptionsStore subs = mock(SubscriptionsStore.class);
m_processor.init(subs, m_storageService, m_sessionStore, null, true);
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator());
//simulate a connect from client1 FAKE_CLIENT_ID from channel1
ConnectMessage connectMessage = new ConnectMessage();
broker/src/test/java/org/eclipse/moquette/spi/impl/FileAuthenticatorTest.java → broker/src/test/java/org/eclipse/moquette/spi/impl/security/FileAuthenticatorTest.java
+1 -2
View File
@@ -13,9 +13,8 @@
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.spi.impl;
package org.eclipse.moquette.spi.impl.security;
import org.eclipse.moquette.server.IAuthenticator;
import org.junit.Test;
import static org.junit.Assert.assertFalse;
parser_commons/src/main/java/org/eclipse/moquette/commons/Constants.java
+2 -1
View File
@@ -36,5 +36,6 @@ public class Constants {
public static final String JKS_PATH_PROPERTY_NAME = "jks_path";
public static final String KEY_STORE_PASSWORD_PROPERTY_NAME = "key_store_password";
public static final String KEY_MANAGER_PASSWORD_PROPERTY_NAME = "key_manager_password";
public static final String ALLOW_ANONYMOUS = "allow_anonymous";
public static final String ALLOW_ANONYMOUS_PROPERTY_NAME = "allow_anonymous";
public static final String ACL_FILE_PROPERTY_NAME = "acl_file";
}