diff --git a/sat/core/sat_main.py b/sat/core/sat_main.py index 33571832..ac96d958 100644 --- a/sat/core/sat_main.py +++ b/sat/core/sat_main.py @@ -939,6 +939,10 @@ class SAT(service.Service): return False return self.profiles[profile].isConnected() + def isAdmin(self, client: xmpp.SatXMPPEntity) -> bool: + """Tells if a client is an administrator with extra privileges""" + return client.profile in self.memory.admins + ## Encryption ## def registerEncryptionPlugin(self, *args, **kwargs): diff --git a/sat/core/xmpp.py b/sat/core/xmpp.py index da5c4e43..94dee804 100644 --- a/sat/core/xmpp.py +++ b/sat/core/xmpp.py @@ -600,6 +600,11 @@ class SatXMPPEntity(core_types.SatXMPPEntity): post_xml_treatments.callback(data) return data + @property + def is_admin(self) -> bool: + """True if a client is an administrator with extra privileges""" + return self.host_app.memory.isAdmin(self.profile) + def addPostXmlCallbacks(self, post_xml_treatments): """Used to add class level callbacks at the end of the workflow @@ -1009,6 +1014,10 @@ class SatXMPPComponent(SatXMPPEntity, component.Component): # FIXME: not the best way to get server jid, maybe use config option? return jid.JID(self.jid.host.split(".", 1)[-1]) + @property + def is_admin(self) -> bool: + return False + def _buildDependencies(self, current, plugins, required=True): """build recursively dependencies needed for a plugin diff --git a/sat/memory/memory.py b/sat/memory/memory.py index afe5ff16..9bc09195 100644 --- a/sat/memory/memory.py +++ b/sat/memory/memory.py @@ -239,6 +239,9 @@ class Memory: self.disco = Discovery(host) self.config = tools_config.parseMainConf(log_filenames=True) self._cache_path = Path(self.getConfig("", "local_dir"), C.CACHE_DIR) + self.admins = self.getConfig("", "admins_list", []) + self.admin_jids = set() + async def initialise(self): self.storage = Storage() @@ -251,6 +254,21 @@ class Memory: self.memory_data = PersistentDict("memory") await self.memory_data.load() await self.disco.load() + for admin in self.admins: + try: + admin_jid_s = await self.asyncGetParamA( + "JabberID", "Connection", profile_key=admin + ) + except Exception as e: + log.warning(f"Can't retrieve jid of admin {admin!r}: {e}") + else: + if admin_jid_s is not None: + try: + admin_jid = jid.JID(admin_jid_s).userhostJID() + except RuntimeError: + log.warning(f"Invalid JID for admin {admin}: {admin_jid_s}") + else: + self.admin_jids.add(admin_jid) ## Configuration ## @@ -1849,3 +1867,16 @@ class Memory: log.debug("No presence information for {}".format(entity_jid)) return False return presence_data.show != C.PRESENCE_UNAVAILABLE + + def isAdmin(self, profile: str) -> bool: + """Tell if given profile has administrator privileges""" + return profile in self.admins + + def isAdminJID(self, entity: jid.JID) -> bool: + """Tells if an entity jid correspond to an admin one + + It is sometime not possible to use the profile alone to check if an entity is an + admin (e.g. a request managed by a component). In this case we check if the JID + correspond to an admin profile + """ + return entity.userhostJID() in self.admin_jids