Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
6 changed files
with
204 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,37 @@ | ||
import asyncio | ||
|
||
|
||
class BaseTopicPlugin: | ||
def __init__(self, context): | ||
self.context = context | ||
try: | ||
self.topic_config = self.context.config['topic-check'] | ||
except KeyError: | ||
self.context.logger.warning("'topic-check' section not found in context configuration") | ||
|
||
def topic_filtering(self, *args, **kwargs): | ||
if not self.topic_config: | ||
# auth config section not found | ||
self.context.logger.warning("'auth' section not found in context configuration") | ||
return False | ||
return True | ||
|
||
|
||
class TopicTabooPlugin(BaseTopicPlugin): | ||
def __init__(self, context): | ||
super().__init__(context) | ||
self._taboo = ['prohibited', 'top-secret', 'data/classified'] | ||
|
||
@asyncio.coroutine | ||
def topic_filtering(self, *args, **kwargs): | ||
filter_result = super().topic_filtering(*args, **kwargs) | ||
if filter_result: | ||
session = kwargs.get('session', None) | ||
topic = kwargs.get('topic', None) | ||
if session.username and topic: | ||
if session.username != 'admin' and topic in self._taboo: | ||
return False | ||
return True | ||
else: | ||
return False | ||
return filter_result |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,49 @@ | ||
import logging | ||
import asyncio | ||
import os | ||
from hbmqtt.broker import Broker | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
config = { | ||
'listeners': { | ||
'default': { | ||
'type': 'tcp', | ||
'bind': '0.0.0.0:1883', | ||
}, | ||
'ws-mqtt': { | ||
'bind': '127.0.0.1:8080', | ||
'type': 'ws', | ||
'max_connections': 10, | ||
}, | ||
}, | ||
'sys_interval': 10, | ||
'auth': { | ||
'allow-anonymous': True, | ||
'password-file': os.path.join(os.path.dirname(os.path.realpath(__file__)), "passwd"), | ||
'plugins': [ | ||
'auth_file', 'auth_anonymous' | ||
] | ||
|
||
}, | ||
'topic-check': { | ||
'enabled': True, | ||
'plugins': [ | ||
'topic_taboo' | ||
] | ||
} | ||
} | ||
|
||
broker = Broker(config) | ||
|
||
|
||
@asyncio.coroutine | ||
def test_coro(): | ||
yield from broker.start() | ||
|
||
|
||
if __name__ == '__main__': | ||
formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" | ||
logging.basicConfig(level=logging.INFO, format=formatter) | ||
asyncio.get_event_loop().run_until_complete(test_coro()) | ||
asyncio.get_event_loop().run_forever() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
import logging | ||
import asyncio | ||
|
||
from hbmqtt.client import MQTTClient, ConnectException | ||
|
||
|
||
# | ||
# This sample shows how to publish messages to broker using different QOS | ||
# Debug outputs shows the message flows | ||
# | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
@asyncio.coroutine | ||
def test_coro(): | ||
try: | ||
C = MQTTClient() | ||
yield from C.connect('mqtt://0.0.0.0:1883') | ||
yield from C.publish('data/classified', b'TOP SECRET', qos=0x01) | ||
yield from C.publish('data/memes', b'REAL FUN', qos=0x01) | ||
logger.info("messages published") | ||
yield from C.disconnect() | ||
except ConnectException as ce: | ||
logger.error("Connection failed: %s" % ce) | ||
asyncio.get_event_loop().stop() | ||
|
||
|
||
if __name__ == '__main__': | ||
formatter = "[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s" | ||
formatter = "%(message)s" | ||
logging.basicConfig(level=logging.DEBUG, format=formatter) | ||
asyncio.get_event_loop().run_until_complete(test_coro()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
import logging | ||
import asyncio | ||
|
||
from hbmqtt.client import MQTTClient, ClientException | ||
from hbmqtt.mqtt.constants import QOS_1 | ||
|
||
|
||
# | ||
# This sample shows how to subscbribe a topic and receive data from incoming messages | ||
# It subscribes to '$SYS/broker/uptime' topic and displays the first ten values returned | ||
# by the broker. | ||
# | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
@asyncio.coroutine | ||
def uptime_coro(): | ||
C = MQTTClient() | ||
yield from C.connect('mqtt://test:test@0.0.0.0:1883') | ||
# Subscribe to '$SYS/broker/uptime' with QOS=1 | ||
yield from C.subscribe([ | ||
('data/memes', QOS_1), # Topic allowed | ||
('data/classified', QOS_1), # Topic forbidden | ||
]) | ||
logger.info("Subscribed") | ||
try: | ||
for i in range(1, 100): | ||
message = yield from C.deliver_message() | ||
packet = message.publish_packet | ||
print("%d: %s => %s" % (i, packet.variable_header.topic_name, str(packet.payload.data))) | ||
yield from C.unsubscribe(['$SYS/broker/uptime', '$SYS/broker/load/#']) | ||
logger.info("UnSubscribed") | ||
yield from C.disconnect() | ||
except ClientException as ce: | ||
logger.error("Client exception: %s" % ce) | ||
|
||
|
||
if __name__ == '__main__': | ||
formatter = "[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s" | ||
logging.basicConfig(level=logging.INFO, format=formatter) | ||
asyncio.get_event_loop().run_until_complete(uptime_coro()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters