Add folder, user and group management

.gitignore

@@ -25,6 +25,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+.venv/
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

passboltapi/__init__.py

@@ -44,6 +44,17 @@ class PassboltError(Exception):
     pass
 
 
+class PassboltGroupNotFoundError(PassboltError):
+    pass
+
+
+class PassboltUserNotFoundError(PassboltError):
+    pass
+
+class PassboltUserNotActiveError(PassboltError):
+    pass
+
+
 class APIClient:
     def __init__(
         self,
@@ -282,7 +293,7 @@ def list_users_with_folder_access(self, folder_id: PassboltFolderIdType) -> List
         # resolve users from groups
         for perm in folder_tuple.permissions:
             if perm.aro == "Group":
-                group_tuple: PassboltGroupTuple = self.describe_group(perm.aro_foreign_key)
+                group_tuple: PassboltGroupTuple = self.describe_group_by_id(perm.aro_foreign_key)
                 for group_user in group_tuple.groups_users:
                     user_ids.add(group_user["user_id"])
             elif perm.aro == "User":
@@ -314,8 +325,9 @@ def import_public_keys(self, trustlevel="TRUST_FULLY"):
         # get all users
         users = self.list_users()
         for user in users:
-            self.gpg.import_keys(user.gpgkey.armored_key)
-            self.gpg.trust_keys(user.gpgkey.fingerprint, trustlevel)
+            if user.gpgkey and user.gpgkey.armored_key:
+                self.gpg.import_keys(user.gpgkey.armored_key)
+                self.gpg.trust_keys(user.gpgkey.fingerprint, trustlevel)
 
     def read_resource(self, resource_id: PassboltResourceIdType) -> PassboltResourceTuple:
         response = self.get(f"/resources/{resource_id}.json", return_response_object=True)
@@ -336,6 +348,48 @@ def read_folder(self, folder_id: PassboltFolderIdType) -> PassboltFolderTuple:
             response["body"]
         )
 
+    def create_folder(self, name: str, folder_id: PassboltFolderIdType) -> PassboltFolderTuple:
+
+        self.read_folder(folder_id=folder_id)
+
+        response = self.post(
+            "/folders.json", {"name": name, "folder_parent_id": folder_id}, return_response_object=True
+        )
+
+        response = response.json()
+        created_folder = constructor(
+            PassboltFolderTuple, subconstructors={"permissions": constructor(PassboltPermissionTuple)})(
+            response["body"]
+        )
+
+        parent_folder = self.read_folder(folder_id)
+
+        # get users with access to parent folder
+        users_list = self.list_users_with_folder_access(folder_id)
+        lookup_users: Mapping[PassboltUserIdType, PassboltUserTuple] = {user.id: user for user in users_list}
+        self_user_id = [user.id for user in users_list if self.user_fingerprint == user.gpgkey.fingerprint]
+        if self_user_id:
+            self_user_id = self_user_id[0]
+        else:
+            raise ValueError("User not in passbolt")
+        # simulate sharing with folder perms
+        permissions = [
+            {
+                "is_new": True,
+                **{k: v for k, v in perm._asdict().items() if k != "id"},
+            }
+            for perm in parent_folder.permissions
+            if (perm.aro_foreign_key != self_user_id)
+        ]
+
+        share_payload = {
+            "permissions": permissions,
+        }
+
+        r_share = self.put(f"/share/folder/{created_folder.id}.json", share_payload, return_response_object=True)
+
+        return created_folder
+
     def describe_folder(self, folder_id: PassboltFolderIdType):
         """Shows folder details with permissions that are needed for some downstream task."""
         response = self.get(
@@ -473,6 +527,156 @@ def update_resource(
             r = self.put(f"/resources/{resource_id}.json", payload, return_response_object=True)
             return r
 
-    def describe_group(self, group_id: PassboltGroupIdType):
+    def describe_user(self, username: str) -> PassboltUserTuple:
+        """
+        Fetch a single user using its username via the read-index endpoint. First search the user using the username.
+        Return a PassboltUserTuple if a user with the exact username provided was found. 
+        Throw PassboltUserNotFoundError instead.
+
+        API Reference : https://help.passbolt.com/api/users/read-index
+        """
+        response = self.get(f"/users.json", params={f"filter[search]": username})
+        found_user = [user for user in response["body"] if user["username"] == username]
+
+        if len(found_user) == 1:
+            return constructor(PassboltUserTuple)(found_user[0])
+        else:
+            raise PassboltUserNotFoundError(f"User {username} not found")
+
+    def describe_user_by_id(self, user_id: PassboltUserIdType) -> PassboltUserTuple:
+        """
+        Fetch a single user using its id.
+        Return a PassboltUserTuple if a user was found. Throw PassboltUserNotFoundError instead.
+
+        API Reference : https://help.passbolt.com/api/users/read
+        """
+        response = self.get(f"/users/{user_id}.json")
+        found_user = response["body"]
+
+        if found_user:
+            return constructor(PassboltUserTuple)(found_user)
+        else:
+            raise PassboltUserNotFoundError(f"User id {user_id} not found")
+
+    def describe_group(self, group_name: str):
+        """
+        Fetch a single group using its name via the read-index endpoint. First list all the groups.
+        Return a PassboltGroupTuple if a group with the exact name provided was found. 
+        Throw PassboltUserNotFoundError instead.
+        """
+        response = self.get(f"/groups.json")
+        found_group = [group for group in response["body"] if group["name"] == group_name]
+
+        if len(found_group) == 1:
+            return constructor(PassboltGroupTuple)(found_group[0])
+        else:
+            raise PassboltGroupNotFoundError(f"Group {group_name} not found")
+
+    def describe_group_by_id(self, group_id: PassboltGroupIdType) -> PassboltGroupTuple:
+        """
+        Fetch a single group using its id.
+        Return a PassboltGroupTuple if the group was found. Throw PassboltUserNotFoundError instead.
+        """
+
         response = self.get(f"/groups/{group_id}.json", params={"contain[groups_users]": 1})
+        found_group = response["body"]
+
+        if found_group:
+            return constructor(PassboltGroupTuple)(found_group)
+        else:
+            raise PassboltGroupNotFoundError(f"Group id {group_id} not found")
+
+    def create_group(self, group_name:str, group_mananger: str) -> PassboltGroupTuple:
+        """
+        Create a group in Passbolt with a user as group manager.
+        Return a PassboltGroupTuple if the group was successfully created.
+
+        API Reference : https://help.passbolt.com/api/groups/create
+        """
+        manager = self.describe_user(group_mananger)
+
+        response = self.post("/groups.json", 
+                             {
+                                 "name": group_name,
+                                 "groups_users": [
+                                        {
+                                            "user_id": manager.id,
+                                            "is_admin": True
+                                        }
+                                 ]
+                             }, return_response_object=True)
+
+        response = response.json()
         return constructor(PassboltGroupTuple)(response["body"])
+
+    def create_user(self, username:str, first_name: str, last_name: str) -> PassboltUserTuple:
+        """
+        Create a user in Passbolt. Return a PassboltUserTuple if the user was sucessfully created
+
+        API Reference : https://help.passbolt.com/api/users/create
+        """
+
+        response = self.post("/users.json", 
+                             {
+                                 "username": username,
+                                 "profile": {
+                                     "first_name": first_name,
+                                     "last_name": last_name
+                                 }
+                            }, return_response_object=True)
+
+        response = response.json()
+        return constructor(PassboltUserTuple)(response["body"])
+
+    def add_user_to_group(self, user_id: PassboltUserIdType, group_id: PassboltGroupIdType) -> PassboltGroupTuple:
+        """
+        Add user to group. User must be active.
+        """
+
+        # Fetch group
+        group = self.describe_group_by_id(group_id=group_id)
+
+        # Fetch user
+        user = self.describe_user_by_id(user_id = user_id)
+
+        if not user.active:
+            raise PassboltUserNotActiveError(f"User {user.username} id {user.id} is inactive : Cannot be added to a grouup")
+
+        # Add user in group
+        user_list = [{
+                "user_id": user.id,
+                "is_admin": False
+            }]
+
+        group_payload =  { "name": group.name, "groups_users": user_list}
+
+        # Update group in API
+        response = self.put(f"/groups/{ group.id }.json", group_payload, return_response_object=True)
+
+        response = response.json()
+        return constructor(PassboltGroupTuple)(response["body"])
+
+    def remove_user_to_group(self, user_id: PassboltUserIdType, group_id: PassboltGroupIdType) -> PassboltGroupTuple:
+        """
+        Remove user from group
+        """
+
+        # Fetch group
+        group = self.describe_group_by_id(group_id=group_id)
+
+        # Fetch user
+        user = self.describe_user_by_id(user_id = user_id)
+
+        # Add user in group
+        group.groups_users.append(
+            {
+                "user_id": user.id,
+                "delete": True
+            }
+        )
+
+        # Update group in API
+        response = self.put(f"/groups/{ group.id }.json", group, return_response_object=True)
+
+        response = response.json()
+        return constructor(PassboltGroupTuple)(response["body"])

requirements.txt

@@ -5,4 +5,4 @@ idna>=2.8
 python-gnupg>=0.4.7
 requests>=2.26.0
 typing-extensions>=4.0.0
-urllib3>=1.25.3
+urllib3>=1.25.3

test.py

@@ -1,7 +1,6 @@
-import time
-
 import passboltapi
 
+from datetime import datetime
 
 def get_my_passwords(passbolt_obj):
     result = list()
@@ -46,13 +45,50 @@ def get_passwords_basic():
 
 
 if __name__ == '__main__':
-    folder_id = "1d932dc0-d0a3-4a44-80c7-4701f84dc307"
-    with passboltapi.PassboltAPI(config_path="config.ini") as passbolt:
-        # required: passbolt.import_public_keys() when the folder has more users.
-        print(passbolt.create_resource(
-            name='Sample Name',
+
+    now = datetime.now() # current date and time
+
+    parent_folder_id = "5b024c98-211c-487e-86f1-0bf75de3f685"
+
+    new_user_name = f"passbolt-py-user-%[email protected]"%(now.strftime("%Y-%m-%d-T%H%M%S"))
+    new_group_name = f"passbolt-py-group-%s"%(now.strftime("%Y-%m-%d-T%H:%M:%S"))
+    new_folder_name = f"passbolt-py-folder-%s"%(now.strftime("%Y-%m-%d-T%H:%M:%S"))
+    new_resource_name = f"passbolt-py-resource-%s"%(now.strftime("%Y-%m-%d-T%H:%M:%S"))
+
+    with passboltapi.PassboltAPI(config_path="config.ini", new_keys=True) as passbolt:
+
+        new_group = passbolt.create_group(new_group_name, "[email protected]")
+        print(f"👏  Created group {new_group.name} with uuid {new_group.id}")
+
+        new_user = passbolt.create_user(
+            username=new_user_name,
+            first_name="Hello",
+            last_name="World from Python API"
+            )
+
+        print(f"👥  Created user {new_user.username} with uuid {new_user.id}")
+
+        # User must be active to be added to a group        
+        # updated_group = passbolt.add_user_to_group(
+        #     user_id=new_user.id,
+        #     group_id=new_group.id
+        # )
+
+        new_folder = passbolt.create_folder(
+            name=new_folder_name,
+            folder_id=parent_folder_id
+        )
+
+        print(f"📂  Created folder {new_folder.name} with uuid {new_folder.id}")
+
+        passbolt.import_public_keys()
+
+        new_resource = passbolt.create_resource(
+            name=new_resource_name,
             username='Sample username',
             password='password_test',
             uri='https://www.passbolt_uri.com',
-            folder_id=folder_id
-        ))
+            folder_id=new_folder.id
+        )
+
+        print(f"🔐  Created ressource {new_resource.name} with uuid {new_resource.id} under {new_folder.name} folder")