From ec49140a7093cc7aebacf9eb5b6cd88c131a7e84 Mon Sep 17 00:00:00 2001 From: jesopo Date: Sun, 17 Feb 2019 14:26:18 +0000 Subject: Move permissions.py in to a directory and add a README.md for it --- modules/permissions.py | 243 ---------------------------------------- modules/permissions/README.md | 32 ++++++ modules/permissions/__init__.py | 243 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 275 insertions(+), 243 deletions(-) delete mode 100644 modules/permissions.py create mode 100644 modules/permissions/README.md create mode 100644 modules/permissions/__init__.py diff --git a/modules/permissions.py b/modules/permissions.py deleted file mode 100644 index e620f46f..00000000 --- a/modules/permissions.py +++ /dev/null @@ -1,243 +0,0 @@ -import base64, os -import scrypt -from src import ModuleManager, utils - -REQUIRES_IDENTIFY = ("You need to be identified to use that command " - "(/msg %s register | /msg %s identify)") - -@utils.export("serverset", {"setting": "identity-mechanism", - "help": "Set the identity mechanism for this server"}) -class Module(ModuleManager.BaseModule): - @utils.hook("new.user") - def new_user(self, event): - self._logout(event["user"]) - - @utils.hook("received.part") - def on_part(self, event): - if len(event["user"].channels) == 1 and event["user" - ].identified_account_override: - event["user"].send_notice("You no longer share any channels " - "with me so you have been signed out") - - def _get_hash(self, server, account): - hash, salt = server.get_user(account).get_setting("authentication", - (None, None)) - return hash, salt - - def _make_salt(self): - return base64.b64encode(os.urandom(64)).decode("utf8") - - def _make_hash(self, password, salt=None): - salt = salt or self._make_salt() - hash = base64.b64encode(scrypt.hash(password, salt)).decode("utf8") - return hash, salt - - def _identified(self, server, user, account): - user.identified_account_override = account - user.identified_account_id_override = server.get_user(account).get_id() - - def _logout(self, user): - user.identified_account_override = None - user.identified_account_id_override = None - - @utils.hook("received.command.identify", private_only=True, min_args=1) - def identify(self, event): - """ - :help: Identify yourself - :usage: [account] - """ - identity_mechanism = event["server"].get_setting("identity-mechanism", - "internal") - if not identity_mechanism == "internal": - raise utils.EventError("The 'identify' command isn't available " - "on this network") - - if not event["user"].channels: - raise utils.EventError("You must share at least one channel " - "with me before you can identify") - - if not event["user"].identified_account_override: - if len(event["args_split"]) > 1: - account = event["args_split"][0] - password = " ".join(event["args_split"][1:]) - else: - account = event["user"].nickname - password = event["args"] - - hash, salt = self._get_hash(event["server"], account) - if hash and salt: - attempt, _ = self._make_hash(password, salt) - if utils.security.constant_time_compare(attempt, hash): - self._identified(event["server"], event["user"], account) - event["stdout"].write("Correct password, you have " - "been identified as '%s'." % account) - self.events.on("internal.identified").call( - user=event["user"]) - else: - event["stderr"].write("Incorrect password for '%s'" % - account) - else: - event["stderr"].write("Account '%s' is not registered" % - account) - else: - event["stderr"].write("You are already identified") - - @utils.hook("received.command.register", private_only=True, min_args=1) - def register(self, event): - """ - :help: Register yourself - :usage: - """ - identity_mechanism = event["server"].get_setting("identity-mechanism", - "internal") - if not identity_mechanism == "internal": - raise utils.EventError("The 'identify' command isn't available " - "on this network") - - hash, salt = self._get_hash(event["server"], event["user"].nickname) - if not hash and not salt: - password = event["args"] - hash, salt = self._make_hash(password) - event["user"].set_setting("authentication", [hash, salt]) - self._identified(event["server"], event["user"], - event["user"].nickname) - event["stdout"].write("Nickname registered successfully") - else: - event["stderr"].write("This nickname is already registered") - - @utils.hook("received.command.setpassword", authenticated=True, min_args=1) - def set_password(self, event): - """ - :help: Change your password - :usage: - """ - hash, salt = self._make_hash(event["args"]) - event["user"].set_setting("authentication", [hash, salt]) - event["stdout"].write("Set your password") - - @utils.hook("received.command.logout", private_only=True) - def logout(self, event): - """ - :help: Logout from your identified account - """ - if event["user"].identified_account_override: - self._logout(event["user"]) - event["stdout"].write("You have been logged out") - else: - event["stderr"].write("You are not logged in") - - @utils.hook("received.command.resetpassword", private_only=True, - min_args=2) - def reset_password(self, event): - """ - :help: Reset a given user's password - :usage: - :permission: resetpassword - """ - target = event["server"].get_user(event["args_split"][0]) - password = " ".join(event["args_split"][1:]) - registered = target.get_setting("authentication", None) - - if registered == None: - event["stderr"].write("'%s' isn't registered" % target.nickname) - else: - hash, salt = self._make_hash(password) - target.set_setting("authentication", [hash, salt]) - event["stdout"].write("Reset password for '%s'" % - target.nickname) - - @utils.hook("preprocess.command") - def preprocess_command(self, event): - permission = event["hook"].get_kwarg("permission", None) - authenticated = event["hook"].kwargs.get("authenticated", False) - - identity_mechanism = event["server"].get_setting("identity-mechanism", - "internal") - identified_account = None - if identity_mechanism == "internal": - identified_account = event["user"].identified_account_override - elif identity_mechanism == "ircv3-account": - identified_account = (event["user"].identified_account or - event["tags"].get("account", None)) - - identified_user = None - permissions = [] - if identified_account: - identified_user = event["server"].get_user(identified_account) - permissions = identified_user.get_setting("permissions", []) - - if permission: - has_permission = permission and ( - permission in permissions or "*" in permissions) - if not identified_account or not has_permission: - return "You do not have permission to do that" - else: - return utils.consts.PERMISSION_FORCE_SUCCESS - elif authenticated: - if not identified_account: - return REQUIRES_IDENTIFY % (event["server"].nickname, - event["server"].nickname) - else: - return utils.consts.PERMISSION_FORCE_SUCCESS - - @utils.hook("received.command.mypermissions", authenticated=True) - def my_permissions(self, event): - """ - :help: Show your permissions - """ - permissions = event["user"].get_setting("permissions", []) - event["stdout"].write("Your permissions: %s" % ", ".join(permissions)) - - def _get_user_details(self, server, nickname): - target = server.get_user(nickname) - registered = bool(target.get_setting("authentication", None)) - permissions = target.get_setting("permissions", []) - return [target, registered, permissions] - - @utils.hook("received.command.givepermission", min_args=2) - def give_permission(self, event): - """ - :help: Give a given permission to a given user - :usage: - :permission: givepermission - """ - permission = event["args_split"][1].lower() - target, registered, permissions = self._get_user_details( - event["server"], event["args_split"][0]) - - if target.get_identified_account() == None: - raise utils.EventError("%s isn't registered" % target.nickname) - - if permission in permissions: - event["stderr"].write("%s already has permission '%s'" % ( - target.nickname, permission)) - else: - permissions.append(permission) - target.set_setting("permissions", permissions) - event["stdout"].write("Gave permission '%s' to %s" % ( - permission, target.nickname)) - @utils.hook("received.command.removepermission", min_args=2) - def remove_permission(self, event): - """ - :help: Remove a given permission from a given user - :usage: - :permission: removepermission - """ - permission = event["args_split"][1].lower() - target, registered, permissions = self._get_user_details( - event["server"], event["args_split"][0]) - - if target.identified_account == None: - raise utils.EventError("%s isn't registered" % target.nickname) - - if permission not in permissions: - event["stderr"].write("%s doesn't have permission '%s'" % ( - target.nickname, permission)) - else: - permissions.remove(permission) - if not permissions: - target.del_setting("permissions") - else: - target.set_setting("permissions", permissions) - event["stdout"].write("Removed permission '%s' from %s" % ( - permission, target.nickname)) diff --git a/modules/permissions/README.md b/modules/permissions/README.md new file mode 100644 index 00000000..18f049d6 --- /dev/null +++ b/modules/permissions/README.md @@ -0,0 +1,32 @@ +# Permissions + +## Adding an admin user + +This is a little complex at the moment but it will get easier some time soon. + +### Registering user + +Join a channel that BitBo is in (he'll automatically join #bitbot with default +configuration) and then type + +> /msg <botnick> register + +### Give * permission + +The `*` permission is a special permission that gives you completely unfettered +access to all of BitBot's functions. + +On IRC, send this to BitBot and take note of the ID response + +> /msg <botnick> myid + +Then take that ID and open the database in sqlite3 (default database location is +`databases/bot.db` + +> $ sqlite3 databases/bot.db + +And then insert your `*` permission + +> INSERT INTO user_settings VALUES (<id>, 'permissions', '["*"]'); + +(where `` is the response from the `myid` command) diff --git a/modules/permissions/__init__.py b/modules/permissions/__init__.py new file mode 100644 index 00000000..e620f46f --- /dev/null +++ b/modules/permissions/__init__.py @@ -0,0 +1,243 @@ +import base64, os +import scrypt +from src import ModuleManager, utils + +REQUIRES_IDENTIFY = ("You need to be identified to use that command " + "(/msg %s register | /msg %s identify)") + +@utils.export("serverset", {"setting": "identity-mechanism", + "help": "Set the identity mechanism for this server"}) +class Module(ModuleManager.BaseModule): + @utils.hook("new.user") + def new_user(self, event): + self._logout(event["user"]) + + @utils.hook("received.part") + def on_part(self, event): + if len(event["user"].channels) == 1 and event["user" + ].identified_account_override: + event["user"].send_notice("You no longer share any channels " + "with me so you have been signed out") + + def _get_hash(self, server, account): + hash, salt = server.get_user(account).get_setting("authentication", + (None, None)) + return hash, salt + + def _make_salt(self): + return base64.b64encode(os.urandom(64)).decode("utf8") + + def _make_hash(self, password, salt=None): + salt = salt or self._make_salt() + hash = base64.b64encode(scrypt.hash(password, salt)).decode("utf8") + return hash, salt + + def _identified(self, server, user, account): + user.identified_account_override = account + user.identified_account_id_override = server.get_user(account).get_id() + + def _logout(self, user): + user.identified_account_override = None + user.identified_account_id_override = None + + @utils.hook("received.command.identify", private_only=True, min_args=1) + def identify(self, event): + """ + :help: Identify yourself + :usage: [account] + """ + identity_mechanism = event["server"].get_setting("identity-mechanism", + "internal") + if not identity_mechanism == "internal": + raise utils.EventError("The 'identify' command isn't available " + "on this network") + + if not event["user"].channels: + raise utils.EventError("You must share at least one channel " + "with me before you can identify") + + if not event["user"].identified_account_override: + if len(event["args_split"]) > 1: + account = event["args_split"][0] + password = " ".join(event["args_split"][1:]) + else: + account = event["user"].nickname + password = event["args"] + + hash, salt = self._get_hash(event["server"], account) + if hash and salt: + attempt, _ = self._make_hash(password, salt) + if utils.security.constant_time_compare(attempt, hash): + self._identified(event["server"], event["user"], account) + event["stdout"].write("Correct password, you have " + "been identified as '%s'." % account) + self.events.on("internal.identified").call( + user=event["user"]) + else: + event["stderr"].write("Incorrect password for '%s'" % + account) + else: + event["stderr"].write("Account '%s' is not registered" % + account) + else: + event["stderr"].write("You are already identified") + + @utils.hook("received.command.register", private_only=True, min_args=1) + def register(self, event): + """ + :help: Register yourself + :usage: + """ + identity_mechanism = event["server"].get_setting("identity-mechanism", + "internal") + if not identity_mechanism == "internal": + raise utils.EventError("The 'identify' command isn't available " + "on this network") + + hash, salt = self._get_hash(event["server"], event["user"].nickname) + if not hash and not salt: + password = event["args"] + hash, salt = self._make_hash(password) + event["user"].set_setting("authentication", [hash, salt]) + self._identified(event["server"], event["user"], + event["user"].nickname) + event["stdout"].write("Nickname registered successfully") + else: + event["stderr"].write("This nickname is already registered") + + @utils.hook("received.command.setpassword", authenticated=True, min_args=1) + def set_password(self, event): + """ + :help: Change your password + :usage: + """ + hash, salt = self._make_hash(event["args"]) + event["user"].set_setting("authentication", [hash, salt]) + event["stdout"].write("Set your password") + + @utils.hook("received.command.logout", private_only=True) + def logout(self, event): + """ + :help: Logout from your identified account + """ + if event["user"].identified_account_override: + self._logout(event["user"]) + event["stdout"].write("You have been logged out") + else: + event["stderr"].write("You are not logged in") + + @utils.hook("received.command.resetpassword", private_only=True, + min_args=2) + def reset_password(self, event): + """ + :help: Reset a given user's password + :usage: + :permission: resetpassword + """ + target = event["server"].get_user(event["args_split"][0]) + password = " ".join(event["args_split"][1:]) + registered = target.get_setting("authentication", None) + + if registered == None: + event["stderr"].write("'%s' isn't registered" % target.nickname) + else: + hash, salt = self._make_hash(password) + target.set_setting("authentication", [hash, salt]) + event["stdout"].write("Reset password for '%s'" % + target.nickname) + + @utils.hook("preprocess.command") + def preprocess_command(self, event): + permission = event["hook"].get_kwarg("permission", None) + authenticated = event["hook"].kwargs.get("authenticated", False) + + identity_mechanism = event["server"].get_setting("identity-mechanism", + "internal") + identified_account = None + if identity_mechanism == "internal": + identified_account = event["user"].identified_account_override + elif identity_mechanism == "ircv3-account": + identified_account = (event["user"].identified_account or + event["tags"].get("account", None)) + + identified_user = None + permissions = [] + if identified_account: + identified_user = event["server"].get_user(identified_account) + permissions = identified_user.get_setting("permissions", []) + + if permission: + has_permission = permission and ( + permission in permissions or "*" in permissions) + if not identified_account or not has_permission: + return "You do not have permission to do that" + else: + return utils.consts.PERMISSION_FORCE_SUCCESS + elif authenticated: + if not identified_account: + return REQUIRES_IDENTIFY % (event["server"].nickname, + event["server"].nickname) + else: + return utils.consts.PERMISSION_FORCE_SUCCESS + + @utils.hook("received.command.mypermissions", authenticated=True) + def my_permissions(self, event): + """ + :help: Show your permissions + """ + permissions = event["user"].get_setting("permissions", []) + event["stdout"].write("Your permissions: %s" % ", ".join(permissions)) + + def _get_user_details(self, server, nickname): + target = server.get_user(nickname) + registered = bool(target.get_setting("authentication", None)) + permissions = target.get_setting("permissions", []) + return [target, registered, permissions] + + @utils.hook("received.command.givepermission", min_args=2) + def give_permission(self, event): + """ + :help: Give a given permission to a given user + :usage: + :permission: givepermission + """ + permission = event["args_split"][1].lower() + target, registered, permissions = self._get_user_details( + event["server"], event["args_split"][0]) + + if target.get_identified_account() == None: + raise utils.EventError("%s isn't registered" % target.nickname) + + if permission in permissions: + event["stderr"].write("%s already has permission '%s'" % ( + target.nickname, permission)) + else: + permissions.append(permission) + target.set_setting("permissions", permissions) + event["stdout"].write("Gave permission '%s' to %s" % ( + permission, target.nickname)) + @utils.hook("received.command.removepermission", min_args=2) + def remove_permission(self, event): + """ + :help: Remove a given permission from a given user + :usage: + :permission: removepermission + """ + permission = event["args_split"][1].lower() + target, registered, permissions = self._get_user_details( + event["server"], event["args_split"][0]) + + if target.identified_account == None: + raise utils.EventError("%s isn't registered" % target.nickname) + + if permission not in permissions: + event["stderr"].write("%s doesn't have permission '%s'" % ( + target.nickname, permission)) + else: + permissions.remove(permission) + if not permissions: + target.del_setting("permissions") + else: + target.set_setting("permissions", permissions) + event["stdout"].write("Removed permission '%s' from %s" % ( + permission, target.nickname)) -- cgit v1.3.1-10-gc9f91