Implement domain management

README.md

@@ -1,12 +1,13 @@
 # Freenit Backend
 ![freenit badge](https://github.com/freenit-framework/backend/actions/workflows/pythonapp.yml/badge.svg)
 
-[Documentation](https://freenit.org/backend/quickstart)
+[Documentation](https://freenit.org/)
 
 [Source](https://github.com/freenit-framework/backend)
 
 Freenit is based on
 
 * [FastAPI](https://fastapi.tiangolo.com/)
 * [Ormar](https://github.com/collerek/ormar)
+* [Bonsai](https://github.com/noirello/bonsai)
 * [Svelte](https://svelte.dev)

freenit/__init__.py

@@ -1 +1 @@
-__version__ = "0.3.16"
+__version__ = "0.3.17"

freenit/api/__init__.py

@@ -1,4 +1,5 @@
 import freenit.api.auth
+import freenit.api.domain
 import freenit.api.role
 import freenit.api.theme
 import freenit.api.user

freenit/api/domain/__init__.py

@@ -0,0 +1,5 @@
+from freenit.models.user import User
+
+if User.dbtype() == "ldap":
+    from .ldap import DomainListAPI, DomainDetailAPI
+

freenit/api/domain/ldap.py

@@ -0,0 +1,59 @@
+import bonsai
+from fastapi import Depends, Header, HTTPException
+
+from freenit.api.router import route
+from freenit.config import getConfig
+from freenit.decorators import description
+from freenit.models.ldap.domain import Domain, DomainCreate
+from freenit.models.pagination import Page
+from freenit.models.user import User
+from freenit.permissions import domain_perms
+
+tags = ["domain"]
+config = getConfig()
+
+
+@route("/domains", tags=tags)
+class DomainListAPI:
+    @staticmethod
+    @description("Get domains")
+    async def get(
+        page: int = Header(default=1),
+        perpage: int = Header(default=10),
+        _: User = Depends(domain_perms),
+    ) -> Page[Domain]:
+        data = await Domain.get_all()
+        perpage = len(data)
+        data = Page(total=perpage, page=page, pages=1, perpage=perpage, data=data)
+        return data
+
+    @staticmethod
+    async def post(data: DomainCreate, _: User = Depends(domain_perms)) -> Domain:
+        if data.name == "":
+            raise HTTPException(status_code=409, detail="Name is mandatory")
+        rdomain, udomain = Domain.create(data.name)
+        try:
+            await rdomain.save()
+            await udomain.save()
+        except bonsai.errors.AlreadyExists:
+            raise HTTPException(status_code=409, detail="Domain already exists")
+        return udomain
+
+
+@route("/domains/{name}", tags=tags)
+class DomainDetailAPI:
+    @staticmethod
+    async def get(name, _: User = Depends(domain_perms)) -> Domain:
+        domain = await Domain.get(name)
+        return domain
+
+    @staticmethod
+    async def delete(name, _: User = Depends(domain_perms)) -> Domain:
+        try:
+            rdomain = await Domain.get_rdomain(name)
+            await rdomain.destroy()
+            domain = await Domain.get(name)
+            await domain.destroy()
+            return domain
+        except bonsai.errors.AuthenticationError:
+            raise HTTPException(status_code=403, detail="Failed to login")

freenit/base_config.py

@@ -64,11 +64,10 @@ def __init__(
         roleBase="dc=group,dc=ldap",
         roleClasses=["groupOfUniqueNames"],
         roleMemberAttr="uniqueMember",
-        groupBase="ou={},dc=group,dc=ldap",
         groupDN="cn={}",
         groupClasses=["posixGroup"],
         userBase="dc=account,dc=ldap",
-        userDN="uid={},ou={}",
+        userDN="uid={}",
         userClasses=["pilotPerson", "posixAccount"],
         userMemberAttr="memberOf",
         uidNextClass="uidNext",
@@ -77,6 +76,8 @@ def __init__(
         gidNextClass="gidNext",
         gidNextDN="cn=gidnext,dc=ldap",
         gidNextField="gidNumber",
+        domainDN="ou={}",
+        domainClasses=["organizationalUnit", "pmiDelegationPath"],
     ):
         self.host = host
         self.tls = tls
@@ -87,9 +88,9 @@ def __init__(
         self.roleDN = f"{roleDN},{roleBase}"
         self.roleMemberAttr = roleMemberAttr
         self.groupClasses = groupClasses
-        self.groupDN = f"{groupDN},{groupBase}"
+        self.groupDN = f"{groupDN},{domainDN},{roleBase}"
         self.userBase = userBase
-        self.userDN = f"{userDN},{userBase}"
+        self.userDN = f"{userDN},{domainDN},{userBase}"
         self.userClasses = userClasses
         self.userMemberAttr = userMemberAttr
         self.uidNextClass = uidNextClass
@@ -98,6 +99,8 @@ def __init__(
         self.gidNextClass = gidNextClass
         self.gidNextDN = gidNextDN
         self.gidNextField = gidNextField
+        self.domainDN = domainDN
+        self.domainClasses = domainClasses
 
 
 class BaseConfig:

freenit/models/ldap/domain.py

@@ -0,0 +1,94 @@
+from bonsai import LDAPEntry, LDAPSearchScope, errors
+from fastapi import HTTPException
+from pydantic import BaseModel, Field
+
+from freenit.config import getConfig
+from freenit.models.ldap.base import LDAPBaseModel, get_client, save_data, class2filter
+
+config = getConfig()
+
+
+class Domain(LDAPBaseModel):
+    ou: str = Field("", description=("Domain name"))
+
+    @classmethod
+    def from_entry(cls, entry):
+        domain = cls(dn=str(entry["dn"]), ou=entry["ou"][0])
+        return domain
+
+    @classmethod
+    def create(cls, fqdn):
+        dn = f"{config.ldap.domainDN},{config.ldap.roleBase}"
+        rdomain = Domain(dn=dn.format(fqdn), ou=fqdn)
+        dn = f"{config.ldap.domainDN},{config.ldap.userBase}"
+        udomain = Domain(dn=dn.format(fqdn), ou=fqdn)
+        return (rdomain, udomain)
+
+    @classmethod
+    async def get(cls, fqdn):
+        classes = class2filter(config.ldap.domainClasses)
+        client = get_client()
+        try:
+            async with client.connect(is_async=True) as conn:
+                dn = f"{config.ldap.domainDN},{config.ldap.userBase}"
+                res = await conn.search(dn.format(fqdn), LDAPSearchScope.SUB, f"(|{classes})")
+        except errors.AuthenticationError:
+            raise HTTPException(status_code=403, detail="Failed to login")
+        if len(res) < 1:
+            raise HTTPException(status_code=404, detail="No such domain")
+        if len(res) > 1:
+            raise HTTPException(status_code=409, detail="Multiple domains found")
+        data = res[0]
+        domain = cls.from_entry(data)
+        return domain
+
+    @classmethod
+    async def get_rdomain(cls, fqdn):
+        classes = class2filter(config.ldap.domainClasses)
+        client = get_client()
+        try:
+            async with client.connect(is_async=True) as conn:
+                dn = f"{config.ldap.domainDN},{config.ldap.roleBase}"
+                res = await conn.search(dn.format(fqdn), LDAPSearchScope.SUB, f"(|{classes})")
+        except errors.AuthenticationError:
+            raise HTTPException(status_code=403, detail="Failed to login")
+        if len(res) < 1:
+            raise HTTPException(status_code=404, detail="No such domain")
+        if len(res) > 1:
+            raise HTTPException(status_code=409, detail="Multiple domains found")
+        data = res[0]
+        domain = cls.from_entry(data)
+        return domain
+
+    @classmethod
+    async def get_all(cls):
+        classes = class2filter(config.ldap.domainClasses)
+        client = get_client()
+        try:
+            async with client.connect(is_async=True) as conn:
+                dn = config.ldap.userBase
+                res = await conn.search(dn, LDAPSearchScope.SUB, f"(|{classes})")
+                data = []
+                for gdata in res:
+                    data.append(cls.from_entry(gdata))
+        except errors.AuthenticationError:
+            raise HTTPException(status_code=403, detail="Failed to login")
+        return data
+
+    async def save(self):
+        data = LDAPEntry(self.dn)
+        data["objectClass"] = config.ldap.domainClasses
+        data["ou"] = self.ou
+        await save_data(data)
+
+    async def destroy(self):
+        client = get_client()
+        try:
+            async with client.connect(is_async=True) as conn:
+                await conn.delete(self.dn)
+        except errors.AuthenticationError:
+            raise HTTPException(status_code=403, detail="Failed to login")
+
+
+class DomainCreate(BaseModel):
+    name: str = Field(description=("Common name"))

freenit/permissions.py

@@ -1,7 +1,8 @@
 from freenit.auth import permissions
 
-role_perms = permissions()
+domain_perms = permissions()
 group_perms = permissions()
 profile_perms = permissions()
-user_perms = permissions()
+role_perms = permissions()
 theme_perms = permissions()
+user_perms = permissions()