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

Implemented the part of parsing and accessing the user specific ACLs

Browse Source
This commit is contained in:
andrea
2015-05-26 00:06:22 +02:00
parent 9a97b00750
commit 059fd89b69
10 changed files with 127 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java
+4 -2
View File
@@ -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<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
public static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
public static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
public static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
@@ -48,7 +50,7 @@ public class NettyChannel implements ServerChannel {
Attribute<Object> 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");
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+14 -9
View File
@@ -215,6 +215,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
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<ValueEvent> {
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<ValueEvent> {
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<ValueEvent> {
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) {
broker/src/main/java/org/eclipse/moquette/spi/impl/security/AuthorizationsCollector.java
+77 -40
View File
@@ -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<Authorization> m_authorizations = new ArrayList();
private List<Authorization> m_globalAuthorizations = new ArrayList();
private List<Authorization> m_patternAuthorizations = new ArrayList();
private Map<String, List<Authorization>> 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<Authorization> 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();
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/DenyAllAuthorizator.java
+2 -2
View File
@@ -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;
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/IAuthorizator.java
+2 -2
View File
@@ -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);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/security/PermitAllAuthorizator.java
+2 -2
View File
@@ -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;
}
}