from litestar.types import ControllerRouterHandler

from .auth.controllers import AuthAdminController, AuthController
from .cidr.controllers import CidrController
from .lists.controllers import ListController
from .web.controllers import WebController, WebPartCidrController, WebPartListController

routes: list[ControllerRouterHandler] = [
    CidrController,
    ListController,
    AuthController,
    AuthAdminController,
    WebController,
    WebPartListController,
    WebPartCidrController,
]

src/app/domain/__init__.py
aorith/cidr-listings
from asyncpg.pool import PoolConnectionProxy
from litestar import Request, Response, post, put
from litestar.controller import Controller
from litestar.datastructures import ResponseHeader, State
from litestar.exceptions import (
    HTTPException,
    InternalServerException,
    NotAuthorizedException,
    ValidationException,
)
from litestar.status_codes import HTTP_200_OK

from app.domain.auth.schemas import (
    Token,
    TokenResponse,
    User,
    UserChangePassword,
    UserLoginOrCreate,
    UserReadDTO,
    UserRoleEnum,
)
from app.domain.auth.services import generate_token
from app.lib.authcrypt import generate_salt_and_hashed_password
from app.lib.settings import get_settings

settings = get_settings()

INSERT_USER = """
INSERT INTO user_login
    (id, login, salt, hashed_password)
VALUES
    ($1, $2, $3, $4);
"""


class AuthAdminController(Controller):
    path = "/v1/admin"
    tags = ["Admin"]

    @post("/signup", return_dto=UserReadDTO)
    async def create_user(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, data: UserLoginOrCreate
    ) -> User:
        """Create user."""
        current_user_record = await conn.fetchrow("select * from user_login where id = $1", request.user.id)
        if not current_user_record:
            raise NotAuthorizedException()
        if current_user_record["role"] != UserRoleEnum.SUPERUSER:
            raise NotAuthorizedException()

        if await conn.fetchval("select login from user_login where login = $1", data.login):
            raise HTTPException(status_code=409, detail="Login name already taken.")
        salt, hashed_password = await generate_salt_and_hashed_password(plain_password=data.password)
        user = User(login=data.login, salt=salt, hashed_password=hashed_password)
        await conn.execute(INSERT_USER, user.id, user.login, user.salt, user.hashed_password)
        return user


class AuthController(Controller):
    path = "/v1/auth"
    tags = ["Auth"]

    @put("/password", status_code=HTTP_200_OK)
    async def change_password(self, request: Request, conn: PoolConnectionProxy, data: UserChangePassword) -> None:
        """Change password."""
        if data.password == data.new_password:
            raise ValidationException("New password must be different.")
        # validate the user generatin a login token with the current password
        user_login = UserLoginOrCreate(login=data.login, password=data.password)
        _ = await generate_token(conn=conn, data=user_login)
        # change the password
        salt, hashed_password = await generate_salt_and_hashed_password(plain_password=data.new_password)
        _id = await conn.fetchval(
            "update user_login set salt = $1, hashed_password = $2 where login = $3 returning id",
            salt,
            hashed_password,
            data.login,
        )
        if not _id:
            raise InternalServerException("Password couldn't be updated.")

    @post(
        "/token",
        status_code=HTTP_200_OK,
        response_headers=[ResponseHeader(name="Authorization", value="Bearer <TOKEN>", description="Token header")],
    )
    async def generate_token(self, conn: PoolConnectionProxy, data: UserLoginOrCreate) -> Response[TokenResponse]:
        """Generate an API access token."""
        token = await generate_token(conn=conn, data=data)
        return Response(
            token,
            headers=[ResponseHeader(name="Authorization", value=f"Bearer {token.access_token}", description="Token")],
        )

src/app/domain/auth/controllers.py
aorith/cidr-listings
from datetime import datetime, timedelta, timezone
from uuid import UUID

from jose import JWTError, jwt
from litestar.exceptions import NotAuthorizedException
from msgspec import to_builtins

from app.domain.auth.schemas import Token, TokenResponse
from app.lib.settings import get_settings

settings = get_settings()


def decode_jwt_token(encoded_token: str) -> Token:
    """Decode JWT token and return ``sub`` value.

    If the token is invalid or expired (i.e. the value stored under the exp key is in the past)
    an exception is raised.
    """
    try:
        payload = jwt.decode(token=encoded_token, key=settings.JWT_SECRET, algorithms=[settings.ALGORITHM])
        return Token(**payload)
    except JWTError as e:
        raise NotAuthorizedException("Invalid token") from e


def encode_jwt_token(user_id: UUID, login: str, expiration: int = settings.DEFAULT_TOKEN_TTL_SECONDS) -> TokenResponse:
    """Encode JWT token with expiration and a given user_id."""
    expires_at = datetime.now(tz=timezone.utc) + timedelta(seconds=expiration)
    token = Token(
        exp=expires_at,
        iat=datetime.now(tz=timezone.utc),
        sub=user_id,
        login=login,
    )
    token_encoded_str = jwt.encode(to_builtins(token), settings.JWT_SECRET, algorithm=settings.ALGORITHM)
    return TokenResponse(access_token=token_encoded_str, expires_in=expiration, expires_at=expires_at)

src/app/domain/auth/jwt.py
aorith/cidr-listings
import msgspec
from litestar.connection import ASGIConnection
from litestar.exceptions import NotAuthorizedException
from litestar.middleware import AbstractAuthenticationMiddleware, AuthenticationResult
from litestar.stores.memory import MemoryStore

from app.domain.auth.jwt import decode_jwt_token
from app.domain.auth.schemas import TokenUser, User
from app.lib.db.base import get_dbmanager
from app.lib.settings import get_settings

settings = get_settings()

store_encoder = msgspec.msgpack.Encoder()
store_decoder = msgspec.msgpack.Decoder(type=TokenUser)
store = MemoryStore()


class JWTAuthenticationMiddleware(AbstractAuthenticationMiddleware):
    async def authenticate_request(self, connection: ASGIConnection) -> AuthenticationResult:
        """Parse the request api key stored in the header and retrieve the user correlating to the token from the DB."""
        if not (auth_header := connection.headers.get("Authorization")):
            if not (cookie_header := connection.cookies.get(settings.API_KEY_COOKIE)):
                raise NotAuthorizedException()
            auth_header = f"Bearer {cookie_header}"

        auth_header = auth_header.strip()
        if not auth_header.startswith("Bearer"):
            raise NotAuthorizedException("Invalid token, should be: 'Bearer <TOKEN>'")
        try:
            auth_header = auth_header.split(" ", 1)[1]
        except IndexError:
            raise NotAuthorizedException(  # noqa: B904
                "Invalid token, should be: 'Bearer <TOKEN>' without extra spaces between 'Bearer' and '<TOKEN>'"
            )

        # try to get a cached token_user pair to avoid a trip to the DB
        if cached_token := await store.get(auth_header):
            token_user = store_decoder.decode(cached_token)
            return AuthenticationResult(user=token_user.user, auth=token_user.token)

        token = decode_jwt_token(encoded_token=auth_header)

        user_record = None
        dbmngr = get_dbmanager()
        async for conn in dbmngr.get_connection():
            user_record = await conn.fetchrow("select * from user_login where id = $1", token.sub)

        if not user_record:
            raise NotAuthorizedException()

        user = User(**user_record)
        token_user = TokenUser(token=token, user=user)
        await store.set(auth_header, store_encoder.encode(token_user), expires_in=settings.AUTH_CACHE_SECONDS)
        return AuthenticationResult(user=user, auth=token)

src/app/domain/auth/middleware.py
aorith/cidr-listings
import uuid
from datetime import datetime
from enum import StrEnum
from typing import Annotated

from litestar.dto import DTOConfig, MsgspecDTO
from msgspec import Meta, Struct, field

from app.lib.default_factories import datetime_no_microseconds

MIN_PLAIN_PASSWORD_LENGTH = 10
MAX_PLAIN_PASSWORD_LENGTH = 64
MIN_LOGIN_LENGTH = 3
MAX_LOGIN_LENGTH = 64


class UserRoleEnum(StrEnum):
    USER = "USER"
    SUPERUSER = "SUPERUSER"


LoginStr = Annotated[
    str,
    Meta(
        min_length=MIN_LOGIN_LENGTH,
        max_length=MAX_LOGIN_LENGTH,
        pattern="^[a-z][a-z0-9_]*$",
        description="Only lowercase letters numbers and '_' allowed, must start with a letter.",
    ),
]

PasswordStr = Annotated[
    str,
    Meta(
        min_length=MIN_PLAIN_PASSWORD_LENGTH,
        max_length=MAX_PLAIN_PASSWORD_LENGTH,
        pattern="(?=.*[0-9])(?=.*[a-z])(?=.*[A-Z]).*",
        description="Minimum ten characters, at least one uppercase letter, one lowercase letter and one number.",
    ),
]


class User(Struct):
    login: LoginStr
    salt: str
    hashed_password: str
    id: uuid.UUID = field(default_factory=uuid.uuid4)  # noqa: A003
    role: UserRoleEnum = UserRoleEnum.USER
    created_at: datetime = field(default_factory=datetime_no_microseconds)
    updated_at: datetime = field(default_factory=datetime_no_microseconds)


class UserLoginOrCreate(Struct):
    login: LoginStr
    password: PasswordStr


class UserChangePassword(UserLoginOrCreate):
    new_password: PasswordStr


class UserReadDTO(MsgspecDTO[User]):
    config = DTOConfig(exclude={"salt", "hashed_password"})


class Token(Struct):
    """Token payload.

    exp: time at which the JWT will expire
    iat: time when the JWT was created
    sub: user/login identification
    """

    exp: datetime | int
    iat: datetime | int
    sub: uuid.UUID
    login: str

    def __post_init__(self) -> None:  # noqa: D105
        if isinstance(self.exp, datetime):
            self.exp = int(self.exp.timestamp())
        if isinstance(self.iat, datetime):
            self.iat = int(self.iat.timestamp())


class TokenResponse(Struct):
    access_token: str
    expires_at: datetime
    expires_in: int
    token_type: str = "Bearer"


class TokenUser(Struct):
    """To cache the middleware response."""

    token: Token
    user: User

src/app/domain/auth/schemas.py
aorith/cidr-listings
from asyncpg.pool import PoolConnectionProxy
from litestar.exceptions import NotAuthorizedException

from app.domain.auth.jwt import encode_jwt_token
from app.domain.auth.schemas import TokenResponse, User, UserLoginOrCreate, UserRoleEnum
from app.lib.authcrypt import generate_salt_and_hashed_password, verify_password
from app.lib.db.base import get_connection

INSERT_USER = """
INSERT INTO user_login
    (id, login, salt, hashed_password, role)
VALUES
    ($1, $2, $3, $4, $5);
"""

UPDATE_USER_PASSWORD = """
UPDATE user_login
SET
    salt = $2, hashed_password = $3
WHERE
    login = $1;
"""  # noqa:S105


async def create_user(login: str, password: str, superuser: bool = False) -> None:
    """Create a new user."""
    async for conn in get_connection():
        async with conn.transaction():
            if await conn.fetchval("select login from user_login where login = $1", login):
                print("Login name already taken.")
                # don't return here to avoid:
                # RuntimeWarning: coroutine 'PoolConnectionHolder.release' was never awaited
            else:
                salt, hashed_password = await generate_salt_and_hashed_password(plain_password=password)
                role = UserRoleEnum.SUPERUSER if superuser else UserRoleEnum.USER

                user = User(login=login, salt=salt, hashed_password=hashed_password, role=role)
                await conn.execute(INSERT_USER, user.id, user.login, user.salt, user.hashed_password, user.role)
                print(user)
                print("User created successfully.")


async def reset_password(login: str, password: str) -> None:
    """Reset user password."""
    async for conn in get_connection():
        async with conn.transaction():
            if not (record := await conn.fetchrow("select * from user_login where login = $1", login)):
                print("Login doesn't exist")
                # don't return here to avoid:
                # RuntimeWarning: coroutine 'PoolConnectionHolder.release' was never awaited
            else:
                salt, hashed_password = await generate_salt_and_hashed_password(plain_password=password)
                user = User(**record)
                user.salt = salt
                user.hashed_password = hashed_password

                await conn.execute(UPDATE_USER_PASSWORD, user.login, user.salt, user.hashed_password)
                print(user)
                print("User password updated successfully.")


async def generate_token(conn: PoolConnectionProxy, data: UserLoginOrCreate) -> TokenResponse:
    """Generate JWT token from an user login."""
    if not (record := await conn.fetchrow("select * from user_login where login = $1", data.login)):
        raise NotAuthorizedException("Wrong login or password.")
    if not await verify_password(
        salt=record["salt"], hashed_password=record["hashed_password"], plain_password=data.password
    ):
        raise NotAuthorizedException("Wrong login or password.")
    return encode_jwt_token(user_id=record["id"], login=record["login"])

src/app/domain/auth/services.py
aorith/cidr-listings
from ipaddress import collapse_addresses, ip_network
from itertools import chain

from asyncpg.pool import PoolConnectionProxy
from litestar import Request, Response
from litestar.controller import Controller
from litestar.datastructures import State
from litestar.handlers import get
from litestar.params import Parameter
from litestar.status_codes import HTTP_200_OK

from app.domain.auth.schemas import Token, User
from app.domain.cidr.schemas import Cidr, CidrByVersion
from app.domain.cidr.services import get_cidr_records
from app.domain.lists.schemas import (
    LIST_ID_PATTERN,
    MAX_LIST_ID_LEN,
    TAG_PARAMS_PATTERN,
    ListTypeEnum,
)


class CidrController(Controller):
    path = "/v1/cidr"
    tags = ["CIDRs"]
    parameters = {
        "list_type": Parameter(ListTypeEnum, description="Required parameter list_type.", required=True),
        "list_id": Parameter(
            str,
            max_length=MAX_LIST_ID_LEN,
            pattern=LIST_ID_PATTERN,
            description="A concrete list_id.",
            required=False,
        ),
        "tags": Parameter(
            str | None,
            description="A tag or multiple tags separated by commas.",
            pattern=TAG_PARAMS_PATTERN,
            required=False,
        ),
    }

    @get("/")
    async def get_cidrs(self, request: Request[User, Token, State], conn: PoolConnectionProxy) -> Response[list[Cidr]]:
        """Get CIDRs from **enabled** lists.

        - The parameter `list_type` is required.

        - If the parameter `list_id` is specified, `tags` filter has no effect
        and if `list_type` doesn't match nothing will be returned.
        """
        records = await get_cidr_records(
            conn=conn,
            user_id=request.user.id,
            list_type=request.query_params["list_type"],
            list_id=request.query_params.get("list_id", None),
            tags=request.query_params.get("tags", None),
        )
        return Response([Cidr(**x) for x in records], status_code=HTTP_200_OK)

    @get("/collapsed")
    async def get_collapsed_cidrs(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy
    ) -> Response[list[str]]:
        """Get CIDRs from **enabled** lists with all the networks collapsed if possible.

        - The parameter `list_type` is required.

        - If the parameter `list_id` is specified, `tags` filter has no effect
        and if `list_type` doesn't match nothing will be returned.
        """
        all_addresses = set()
        for record in await get_cidr_records(
            conn=conn,
            user_id=request.user.id,
            list_type=request.query_params["list_type"],
            list_id=request.query_params.get("list_id", None),
            tags=request.query_params.get("tags", None),
        ):
            all_addresses.add(ip_network(record["address"]))

        ipv4_collapsed = collapse_addresses(x for x in all_addresses if x.version == 4)
        ipv6_collapsed = collapse_addresses(x for x in all_addresses if x.version == 6)
        return Response([x.compressed for x in chain(ipv4_collapsed, ipv6_collapsed)])

    @get("/collapsed/by-ip-version")
    async def get_collapsed_by_version_cidrs(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy
    ) -> Response[CidrByVersion]:
        """Get CIDRs from **enabled** lists, collapsed by ip version.

        - The parameter `list_type` is required.

        - If the parameter `list_id` is specified, `tags` filter has no effect
        and if `list_type` doesn't match nothing will be returned.
        """
        all_addresses = set()
        for record in await get_cidr_records(
            conn=conn,
            user_id=request.user.id,
            list_type=request.query_params["list_type"],
            list_id=request.query_params.get("list_id", None),
            tags=request.query_params.get("tags", None),
        ):
            all_addresses.add(ip_network(record["address"]))

        ipv4_collapsed = collapse_addresses(x for x in all_addresses if x.version == 4)
        ipv6_collapsed = collapse_addresses(x for x in all_addresses if x.version == 6)
        return Response(CidrByVersion(ipv4=list(ipv4_collapsed), ipv6=list(ipv6_collapsed)))

src/app/domain/cidr/controllers.py
aorith/cidr-listings
from datetime import datetime

from msgspec import Struct, field

from app.lib.default_factories import datetime_no_microseconds


class Cidr(Struct):
    address: str
    list_id: str
    expires_at: datetime | None = None
    created_at: datetime = field(default_factory=datetime_no_microseconds)
    updated_at: datetime = field(default_factory=datetime_no_microseconds)
    id: int | None = None


class CidrNL(Struct):
    """Same as Cidr but without list_id.

    for /v1/list/<id>/cidr.
    """

    address: str
    expires_at: datetime
    created_at: datetime
    updated_at: datetime


class CidrByVersion(Struct):
    ipv4: list[str]
    ipv6: list[str]

src/app/domain/cidr/schemas.py
aorith/cidr-listings
from uuid import UUID

from asyncpg import Record
from asyncpg.pool import PoolConnectionProxy
from litestar.exceptions import ImproperlyConfiguredException

SELECT_ENABLED_BY_TYPE_AND_ID = """
select *
from cidr
where list_id
in
(select id
 from list
 where
 enabled = true
 and user_id = $1
 and list_type = $2
 and list_id = $3
)"""

SELECT_ENABLED_BY_TYPE_AND_TAGS = """
select *
from cidr
where list_id
in
(select id
 from list
 where
 enabled = true
 and user_id = $1
 and list_type = $2
 and tags && $3::text[]
)"""

SELECT_ENABLED_BY_ID = """
select *
from cidr
where list_id
in
(select id
 from list
 where
 enabled = true
 and user_id = $1
 and list_id = $2
)"""

SELECT_ENABLED_BY_TYPE = """
select *
from cidr
where
list_id in
(select id
 from list
 where
 enabled = true
 and user_id = $1
 and list_type = $2
)"""

SELECT_BY_ID_FIRST_PAGE = """
select *
from cidr
where
list_id = $1
order by id desc
limit $2
"""

SELECT_BY_ID_PAGINATED = """
select *
from cidr
where
id < $1
and
list_id = $2
order by id desc
limit $3
"""


async def get_cidr_records(
    conn: PoolConnectionProxy,
    user_id: UUID,
    list_type: str | None = None,
    list_id: str | None = None,
    tags: str | None = None,
) -> list[Record]:
    """Get CIDR records filtering by id, type and or tags."""
    if list_id and list_type:
        return await conn.fetch(
            SELECT_ENABLED_BY_TYPE_AND_ID,
            user_id,
            list_type,
            list_id,
        )

    if tags and list_type:
        tags_split = [x.strip() for x in tags.strip().split(",") if x]
        return await conn.fetch(
            SELECT_ENABLED_BY_TYPE_AND_TAGS,
            user_id,
            list_type,
            tags_split,
        )

    if list_id:
        return await conn.fetch(
            SELECT_ENABLED_BY_ID,
            user_id,
            list_id,
        )

    if not list_type:
        raise ImproperlyConfiguredException("list_type is mandatory if no other filters are used")

    return await conn.fetch(
        SELECT_ENABLED_BY_TYPE,
        user_id,
        list_type,
    )


async def get_cidr_records_paginated(
    conn: PoolConnectionProxy,
    list_id: str,
    address_id: int | None = None,
    limit: int = 100,
) -> list[Record]:
    """Get CIDR records paginating by id."""
    if address_id:
        return await conn.fetch(
            SELECT_BY_ID_PAGINATED,
            address_id,
            list_id,
            limit,
        )

    return await conn.fetch(
        SELECT_BY_ID_FIRST_PAGE,
        list_id,
        limit,
    )

src/app/domain/cidr/services.py
aorith/cidr-listings
from asyncpg.pool import PoolConnectionProxy
from litestar import Request, Response
from litestar.controller import Controller
from litestar.datastructures import State
from litestar.dto import DTOData
from litestar.exceptions import HTTPException, NotFoundException
from litestar.handlers import delete, get, post, put
from litestar.status_codes import HTTP_400_BAD_REQUEST

from app.domain.auth.schemas import Token, User
from app.domain.cidr.schemas import CidrNL
from app.domain.lists.schemas import (
    ActionEnum,
    CidrAdd,
    CidrAddRaw,
    CidrDelete,
    CidrDeleteRaw,
    CidrJob,
    CidrList,
    ListCreateDTO,
    ListFull,
    ListTypeEnum,
    ListUpdateDTO,
)
from app.domain.lists.services import insert_cidr_job, parse_raw_cidrs_input_as_str
from app.lib.validations import run_validation

INSERT_LIST = """
INSERT INTO list
    (id, user_id, list_type, enabled, tags, description)
VALUES
    ($1, $2, $3, $4, $5, $6)
ON CONFLICT (id)
    DO NOTHING
RETURNING *;

-- Returns the id if it was created successfully, else nothing
"""
UPDATE_LIST = """
UPDATE list
SET
    list_type = $1,
    enabled = $2,
    tags = $3::text[],
    description = $4
WHERE
    id = $5 and user_id = $6
RETURNING *;
"""


class ListController(Controller):
    path = "/v1/list"
    tags = ["Lists"]

    @get("/", dto=None)
    async def get_lists(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy
    ) -> Response[list[ListFull]]:
        """Get lists."""
        records = await conn.fetch("select * from list where user_id = $1", request.user.id)
        return Response([ListFull(**x) for x in records])

    @post("/", dto=ListCreateDTO, return_dto=None)
    async def create_list(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, data: ListFull
    ) -> Response[ListFull]:
        """Create list."""
        await run_validation(data=data, target_type=ListFull)
        async with conn.transaction():
            record = await conn.fetchrow(
                INSERT_LIST,
                data.id,
                request.user.id,
                data.list_type,
                data.enabled,
                list({*data.tags, "DEFAULT"}),
                data.description,
            )
        if not record:
            raise HTTPException(status_code=409, detail=f"List {data.id} already exists.")
        return Response(ListFull(**record))

    @get("/{id:str}")
    async def get_list(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str
    ) -> Response[ListFull]:
        """Get list."""
        record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not record:
            raise NotFoundException(f"List {id} not found")
        return Response(ListFull(**record))

    @put("/{id:str}", dto=ListUpdateDTO, return_dto=None)
    async def update_list(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str, data: DTOData[ListFull]
    ) -> Response[ListFull]:
        """Update list."""
        to_update = data.as_builtins()
        async with conn.transaction():
            current_record = await conn.fetchrow(
                "select * from list where id = $1 and user_id = $2", id, request.user.id
            )
            if not current_record:
                raise NotFoundException(f"List {id} not found.")

            list_type = to_update.get("list_type", current_record["list_type"])
            enabled = to_update.get("enabled", current_record["enabled"])
            tags = to_update.get("tags", current_record["tags"])
            description = to_update.get("description", current_record["description"])

            # Force validation
            await run_validation(
                data=ListFull(
                    id=id,
                    user_id=request.user.id,
                    list_type=list_type,
                    enabled=enabled,
                    tags=list({*tags, "DEFAULT"}),
                    description=description,
                ),
                target_type=ListFull,
            )

            record = await conn.fetchrow(
                UPDATE_LIST, list_type, enabled, list({*tags, "DEFAULT"}), description, id, request.user.id
            )
            if not record:
                raise NotFoundException(f"List {id} not found.")

            # force deny cidrs cleanup if the list is safe and goes from disabled to enabled
            if list_type == ListTypeEnum.SAFE and not current_record["enabled"] and to_update.get("enabled", False):
                cidr_job = CidrJob(
                    action=ActionEnum.UPDATE,
                    list_id=id,
                    list_type=list_type,
                    list_enabled=enabled,
                    user_id=request.user.id,
                    cidrs=[],
                    ttl=None,
                )
                await insert_cidr_job(job=cidr_job, conn=conn)

            return Response(ListFull(**record))

    @delete("/{id:str}")
    async def delete_list(self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str) -> None:
        """Delete list."""
        async with conn.transaction():
            record = await conn.fetchval(
                "delete from list where id = $1 and user_id = $2 returning id", id, request.user.id
            )
            if not record:
                raise NotFoundException(f"List {id} not found.")

    @get("/{id:str}/cidr")
    async def get_cidrs(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str
    ) -> Response[CidrList]:
        """Get CIDRs."""
        list_record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not list_record:
            raise NotFoundException(f"List {id} not found.")
        records = await conn.fetch(  # even if we have the list_id, we need to ensure that the list belongs to the user
            "select * from cidr where list_id = (select id from list where id = $1 and user_id = $2)",
            id,
            request.user.id,
        )
        cidrs = [
            CidrNL(
                address=x["address"], expires_at=x["expires_at"], created_at=x["created_at"], updated_at=x["updated_at"]
            )
            for x in records
        ]
        return Response(CidrList(cidrs=cidrs, **list_record))

    @post("/{id:str}/cidr/add")
    async def add_cidrs(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str, data: CidrAdd
    ) -> Response[CidrJob]:
        """Create a job to add CIDRs.

        Validation of CIDRs is performed asynchronously.

        - CIDRs from the non-routable address space are discarded automatically if
        the server side environment variable `ONLY_GLOBAL_CIDRS` is `True` (which is by default).

        - If a CIDR already exist in the list its `ttl` will be updated.

        - If the `list_type` is `SAFE` another job will delete all the matching CIDRs from lists of type `DENY`.
        """
        if data.ttl is not None and data.ttl <= 0:
            raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="TTL must be greater than 0.")

        list_record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not list_record:
            raise NotFoundException(f"List {id} not found.")

        cidr_job = CidrJob(
            action=ActionEnum.ADD,
            list_id=list_record["id"],
            list_type=list_record["list_type"],
            list_enabled=list_record["enabled"],
            user_id=request.user.id,
            cidrs=data.cidrs,
            ttl=data.ttl,
        )

        await insert_cidr_job(job=cidr_job, conn=conn)
        return Response(cidr_job)

    @post("/{id:str}/cidr/delete")
    async def delete_cidrs(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str, data: CidrDelete
    ) -> Response[CidrJob]:
        """Create a job to delete CIDRs.

        Validation of CIDRs is performed asynchronously.
        """
        list_record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not list_record:
            raise NotFoundException(f"List {id} not found.")

        cidr_job = CidrJob(
            action=ActionEnum.DELETE,
            list_id=list_record["id"],
            list_type=list_record["list_type"],
            list_enabled=list_record["enabled"],
            user_id=request.user.id,
            cidrs=data.cidrs,
            ttl=None,
        )

        await insert_cidr_job(job=cidr_job, conn=conn)
        return Response(cidr_job)

    @post("/{id:str}/cidr/add/raw")
    async def add_cidrs_raw(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str, data: CidrAddRaw
    ) -> Response[CidrJob]:
        """Create a job to add CIDRs.

        The string sent will be parsed in search of valid ipv4/ipv6 CIDRs.

        Validation of CIDRs is performed asynchronously.

        - CIDRs from the non-routable address space are discarded automatically if
        the server side environment variable `ONLY_GLOBAL_CIDRS` is `True` (which is by default).

        - If a CIDR already exist in the list its `ttl` will be updated.

        - If the `list_type` is `SAFE` another job will delete all the matching CIDRs from lists of type `DENY`.
        """
        if data.ttl is not None and data.ttl <= 0:
            raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="TTL must be greater than 0.")

        list_record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not list_record:
            raise NotFoundException(f"List {id} not found.")

        ipv4_cidrs, ipv6_cidrs = parse_raw_cidrs_input_as_str(raw_data=data.cidrs)
        cidr_job = CidrJob(
            action=ActionEnum.ADD,
            list_id=list_record["id"],
            list_type=list_record["list_type"],
            list_enabled=list_record["enabled"],
            user_id=request.user.id,
            cidrs=list(ipv4_cidrs | ipv6_cidrs),
            ttl=data.ttl,
        )

        await insert_cidr_job(job=cidr_job, conn=conn)
        return Response(cidr_job)

    @post("/{id:str}/cidr/delete/raw")
    async def delete_cidrs_raw(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str, data: CidrDeleteRaw
    ) -> Response[CidrJob]:
        """Create a job to delete CIDRs.

        The string sent will be parsed in search of valid ipv4/ipv6 CIDRs.

        Validation of CIDRs is performed asynchronously.
        """
        list_record = await conn.fetchrow("select * from list where id = $1 and user_id = $2", id, request.user.id)
        if not list_record:
            raise NotFoundException(f"List {id} not found.")

        ipv4_cidrs, ipv6_cidrs = parse_raw_cidrs_input_as_str(raw_data=data.cidrs)
        cidr_job = CidrJob(
            action=ActionEnum.DELETE,
            list_id=list_record["id"],
            list_type=list_record["list_type"],
            list_enabled=list_record["enabled"],
            user_id=request.user.id,
            cidrs=list(ipv4_cidrs | ipv6_cidrs),
            ttl=None,
        )

        await insert_cidr_job(job=cidr_job, conn=conn)
        return Response(cidr_job)

src/app/domain/lists/controllers.py
aorith/cidr-listings
import uuid
from datetime import datetime
from enum import StrEnum
from typing import Annotated

from litestar.dto import DTOConfig, MsgspecDTO
from msgspec import Meta, Struct, field

from app.domain.cidr.schemas import CidrNL
from app.lib.default_factories import datetime_no_microseconds

MAX_LIST_ID_LEN = 64
MAX_TAG_LEN = 16
MAX_DESCRIPTION_LEN = 256

LIST_ID_PATTERN = "^[A-Z][A-Z0-9_]*$"
TAG_PATTERN = "^[A-Z][A-Z0-9]*$"
TAG_PARAMS_PATTERN = "^([A-Z][A-Z0-9]*,?)+$"

CidrType = Annotated[str, Meta(description="IP address or network in CIDR notation.")]

CidrTTL = Annotated[
    int | None,
    Meta(
        description="Time to live (TTL) of each CIDR in seconds, after that they'll beremoved from the list, when this field is not set CIDRs won't be removed."
    ),
]

TagStr = Annotated[
    str,
    Meta(
        max_length=MAX_TAG_LEN,
        pattern=TAG_PATTERN,
        description="Uppercase letters and numbers, starting with a letter. The 'DEFAULT' tag will be always present.",
    ),
]


def default_tag_factory() -> list[TagStr]:
    return ["DEFAULT"]


class ListTypeEnum(StrEnum):
    DENY = "DENY"
    SAFE = "SAFE"


class ActionEnum(StrEnum):
    ADD = "add"
    DELETE = "delete"
    UPDATE = "update"


class ListBase(Struct):
    id: Annotated[
        str,
        Meta(
            max_length=MAX_LIST_ID_LEN,
            pattern=LIST_ID_PATTERN,
            description="Uppercase letters, numbers and underscores (`_`) allowed, must start with a letter.",
        ),
    ]
    list_type: ListTypeEnum
    tags: list[TagStr] = field(default_factory=default_tag_factory)
    enabled: bool = False
    description: Annotated[str, Meta(max_length=MAX_DESCRIPTION_LEN)] = field(default="")


class ListFull(ListBase):
    user_id: uuid.UUID = field(default_factory=uuid.uuid4)
    created_at: datetime = field(default_factory=datetime_no_microseconds)
    updated_at: datetime = field(default_factory=datetime_no_microseconds)


class ListCreateDTO(MsgspecDTO[ListFull]):
    config = DTOConfig(exclude={"user_id", "created_at", "updated_at"})


class ListUpdateDTO(MsgspecDTO[ListFull]):
    config = DTOConfig(partial=True, exclude={"id", "user_id", "created_at", "updated_at"})


class CidrList(ListFull):
    cidrs: list[CidrNL] = field(default_factory=list)


class CidrDelete(Struct):
    cidrs: list[CidrType]


class CidrDeleteRaw(Struct):
    cidrs: str


class CidrAdd(CidrDelete):
    cidrs: list[CidrType]
    ttl: CidrTTL = None


class CidrAddRaw(CidrDelete):
    cidrs: str
    ttl: CidrTTL = None


class CidrJob(Struct):
    list_id: str
    list_type: ListTypeEnum
    list_enabled: bool
    user_id: uuid.UUID
    action: ActionEnum
    cidrs: list[str]
    ttl: CidrTTL = None
    job_id: uuid.UUID = field(default_factory=uuid.uuid4)

src/app/domain/lists/schemas.py
aorith/cidr-listings
import itertools
import re
from ipaddress import IPv4Network, IPv6Network, ip_network

import msgspec
from asyncpg.pool import PoolConnectionProxy

from app.domain.lists.schemas import CidrJob

INSERT_JOB = """
INSERT INTO job_queue
    (job_id, payload)
VALUES
    ($1, $2::jsonb)
"""


IPV4_RE = re.compile(r"\b((?:[0-9]{1,3}\.){3}[0-9]{1,3}(?:\/[0-9]{1,2})?)\b")
IPV6_RE = re.compile(r"\b([A-Fa-f0-9:]+:[A-Fa-f0-9]*(?:\/[0-9]{1,3})?)\b")

json_enc = msgspec.json.Encoder()


def parse_raw_cidrs_input(raw_data: str) -> tuple[set[IPv4Network], set[IPv6Network]]:
    """Parse raw input string and return valid IPv4/IPv6 networks."""
    ipv4_valid: set[IPv4Network] = set()
    ipv6_valid: set[IPv6Network] = set()

    for i in itertools.chain(re.finditer(IPV4_RE, raw_data), re.finditer(IPV6_RE, raw_data)):
        try:
            ipn = ip_network(i.group(0), strict=False)
            if ipn.version == 4:
                ipv4_valid.add(ipn)
            else:
                ipv6_valid.add(ipn)
        except ValueError:
            continue

    return ipv4_valid, ipv6_valid


def parse_raw_cidrs_input_as_str(raw_data: str) -> tuple[set[str], set[str]]:
    """Parse raw input string and return valid IPv4/IPv6 networks as a string."""
    ipv4_valid, ipv6_valid = parse_raw_cidrs_input(raw_data=raw_data)
    return {x.compressed for x in ipv4_valid}, {x.compressed for x in ipv6_valid}


async def insert_cidr_job(job: CidrJob, conn: PoolConnectionProxy) -> None:
    """Insert a new CIDR Job."""
    cidr_job_json = json_enc.encode(job).decode()
    async with conn.transaction():
        await conn.execute(
            INSERT_JOB,
            job.job_id,
            cidr_job_json,
        )

src/app/domain/lists/services.py
aorith/cidr-listings
import ipaddress
import re
from typing import Annotated
from uuid import UUID

import msgspec
from asyncpg.pool import PoolConnectionProxy
from litestar import Request, Response, delete, get, post, put
from litestar.controller import Controller
from litestar.datastructures import Cookie, State
from litestar.enums import RequestEncodingType
from litestar.exceptions import HTTPException, NotFoundException, ValidationException
from litestar.params import Body
from litestar.response import Template
from litestar.status_codes import HTTP_200_OK, HTTP_204_NO_CONTENT
from msgspec import ValidationError

from app.domain.auth.schemas import Token, User, UserLoginOrCreate
from app.domain.auth.services import generate_token
from app.domain.cidr.services import get_cidr_records_paginated
from app.domain.lists.controllers import INSERT_LIST, UPDATE_LIST
from app.domain.lists.schemas import ActionEnum, CidrJob, ListFull, ListTypeEnum
from app.domain.lists.services import insert_cidr_job, parse_raw_cidrs_input, parse_raw_cidrs_input_as_str
from app.lib.settings import get_settings

settings = get_settings()


SELECT_LISTS_WITH_CIDR_COUNT = """
select l.id, coalesce(c1.networks, 0) as networks, l.list_type, l.enabled, l.tags, l.description, l.updated_at
from list l
left join
    (
        select c.list_id, count(c.address) as networks from
        cidr c
        group by c.list_id
    ) c1
on l.id = c1.list_id
where l.user_id = $1
order by l.updated_at desc
"""

SELECT_LIST_WITH_CIDR_COUNT = """
select l.id, coalesce(c1.networks, 0) as networks, l.list_type, l.enabled, l.tags, l.description, l.updated_at
from list l
left join
    (
        select c.list_id, count(c.address) as networks from
        cidr c
        group by c.list_id
    ) c1
on l.id = c1.list_id
where l.user_id = $1
and l.id = $2
order by l.updated_at desc
"""


async def get_network_info(address: str | None, conn: PoolConnectionProxy, user_id: UUID) -> Template:
    """Get network info."""
    error = None
    ipn = None
    usable = None
    subnets = None
    supernet = None
    if address:
        try:
            ipn = ipaddress.ip_network(address)
            usable = max(ipn.num_addresses - 2, 1)
            subnets = ipn.subnets()
            supernet = ipn.supernet()
        except ValueError as err:
            error = str(err)

    lists = [
        x["id"]
        for x in await conn.fetch(
            "select id from list where user_id = $1 and id in (select list_id from cidr where address >>= $2)",
            user_id,
            ipn,
        )
    ]

    return Template(
        template_name="partials/network-info.html.j2",
        context={"error": error, "usable": usable, "ip": ipn, "subnets": subnets, "supernet": supernet, "lists": lists},
    )


class WebController(Controller):
    tags = ["Web"]

    @get(["/", "/lists", "/cidrs", "/network-info"], include_in_schema=False)
    async def home(self, request: Request[User, Token, State]) -> Template:
        """Home."""
        hx_get = None
        hx_push_url = None
        if request.url.path in ["/", "/lists"]:
            hx_get = "/parts/list/"
            hx_push_url = "/lists"
        elif request.url.path == "/cidrs":
            hx_get = "/parts/cidr"
            hx_push_url = request.url.path
        elif request.url.path == "/network-info":
            hx_get = "/parts/cidr/ni"
            hx_push_url = request.url.path
        return Template(
            template_name="base.html.j2",
            context={"login": request.user.login, "hx_get": hx_get, "hx_push_url": hx_push_url},
        )

    @get("/list/{id:str}", include_in_schema=False)
    async def home_list(self, request: Request[User, Token, State], id: str) -> Template:
        """Home/list."""
        return Template(
            template_name="base.html.j2",
            context={"login": request.user.login, "hx_get": f"/parts/list/{id}", "hx_push_url": request.url.path},
        )

    @get("/cidr/{ip:str}/{prefix:str}", include_in_schema=False)
    async def home_cidr(self, request: Request[User, Token, State], ip: str, prefix: str) -> Template:
        """Home/cidr."""
        return Template(
            template_name="base.html.j2",
            context={
                "login": request.user.login,
                "hx_get": f"/parts/cidr/ni/{ip}/{prefix}",
                "hx_push_url": request.url.path,
            },
        )

    @get(path="/login", include_in_schema=False)
    async def login(self) -> Template:
        """Login."""
        return Template(template_name="login.html.j2", context={})

    @get(path="/logout", include_in_schema=False)
    async def logout(self) -> Response:
        """Logout."""
        return Response(
            None,
            headers={"HX-Push": "/", "HX-Redirect": "/"},
            cookies=[Cookie(key=settings.API_KEY_COOKIE, value="", samesite="strict", max_age=-1)],
            status_code=HTTP_204_NO_CONTENT,
        )

    @post("/login", include_in_schema=False)
    async def post_login(
        self,
        conn: PoolConnectionProxy,
        data: Annotated[UserLoginOrCreate, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Response:
        """Login."""
        token = await generate_token(conn=conn, data=data)
        return Response(
            None,
            headers={"HX-Push": "/", "HX-Redirect": "/"},
            cookies=[
                Cookie(
                    key=settings.API_KEY_COOKIE, value=token.access_token, samesite="strict", max_age=token.expires_in
                )
            ],
            status_code=HTTP_204_NO_CONTENT,
        )


class WebPartListController(Controller):
    tags = ["Web"]
    path = "/parts/list"

    @get("/", include_in_schema=False)
    async def get_lists(self, request: Request[User, Token, State], conn: PoolConnectionProxy) -> Template:
        """Get lists."""
        records = await conn.fetch(SELECT_LISTS_WITH_CIDR_COUNT, request.user.id)
        return Template(template_name="partials/lists.html.j2", context={"items": records})

    @post("/", include_in_schema=False)
    async def create_list(
        self,
        request: Request[User, Token, State],
        conn: PoolConnectionProxy,
        data: Annotated[dict, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Response:
        """Create list."""
        tags = {x.strip() for x in re.split(",| |\n", data["tags"]) if x}
        tags.add("DEFAULT")
        data.update(
            {
                "enabled": data.get("enabled", "false").lower() in ["true", "on"],
                "tags": list(tags),
                "user_id": UUID(
                    str(request.user.id),
                ),
            }
        )
        try:
            new_list = msgspec.convert(data, ListFull)
        except ValidationError as err:
            raise ValidationException(str(err))  # noqa: B904

        async with conn.transaction():
            record = await conn.fetchrow(
                INSERT_LIST,
                new_list.id,
                request.user.id,
                new_list.list_type,
                new_list.enabled,
                new_list.tags,
                new_list.description,
            )
        if not record:
            raise HTTPException(status_code=409, detail=f"List {new_list.id} already exists.")

        return Template(template_name="partials/list-item.html.j2", context={"item": record, "new_item": True})

    @get("/{id:str}", include_in_schema=False)
    async def get_list_detail(self, conn: PoolConnectionProxy, id: str, last_cidr_id: int | None = None) -> Template:
        """Get list detail."""
        if last_cidr_id:
            records = await get_cidr_records_paginated(conn=conn, list_id=id, address_id=last_cidr_id)
            if len(records) > 0:
                last_cidr_id = records[-1]["id"]
            return Template(
                template_name="partials/cidr-detail.html.j2",
                context={"list_id": id, "items": records, "last_cidr_id": last_cidr_id},
            )

        records = await get_cidr_records_paginated(conn=conn, list_id=id)
        if len(records) > 0:
            last_cidr_id = records[-1]["id"]
        return Template(
            template_name="partials/list-cidrs.html.j2",
            context={"list_id": id, "items": records, "last_cidr_id": last_cidr_id},
        )

    @put("/{id:str}", include_in_schema=False)
    async def edit_list(
        self,
        request: Request[User, Token, State],
        conn: PoolConnectionProxy,
        id: str,
        data: Annotated[dict, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Template:
        """Edit list."""
        tags = {x.strip() for x in re.split(",| |\n", data["tags"]) if x}
        tags.add("DEFAULT")
        data.update(
            {
                "id": id,
                "list_type": data["list_type"],
                "user_id": request.user.id,
                "enabled": data.get("enabled", "false").lower() in ["true", "on"],
                "tags": list(tags),
                "description": data["description"],
            }
        )
        try:
            updated_list = msgspec.convert(data, ListFull)
        except ValidationError as err:
            raise ValidationException(str(err))  # noqa: B904

        async with conn.transaction():
            record = await conn.fetchrow(
                "select id, enabled from list where id = $1 and user_id = $2", id, request.user.id
            )
            if not record:
                raise NotFoundException(f"List {id} not found.")

            # force deny cidrs cleanup if the list is safe and goes from disabled to enabled
            if data["list_type"] == ListTypeEnum.SAFE and not record["enabled"] and data.get("enabled", False):
                cidr_job = CidrJob(
                    action=ActionEnum.UPDATE,
                    list_id=id,
                    list_type=data["list_type"],
                    list_enabled=data["enabled"],
                    user_id=request.user.id,
                    cidrs=[],
                    ttl=None,
                )
                await insert_cidr_job(job=cidr_job, conn=conn)

            record = await conn.fetchrow(
                UPDATE_LIST,
                updated_list.list_type,
                updated_list.enabled,
                updated_list.tags,
                updated_list.description,
                id,
                request.user.id,
            )

        records = await conn.fetch(SELECT_LIST_WITH_CIDR_COUNT, request.user.id, id)
        if not records:
            raise NotFoundException(f"List {id} not found.")
        item = dict(records[0])

        return Template(template_name="partials/list-item.html.j2", context={"new_item": True, "item": item})

    @get("/edit/{id:str}", include_in_schema=False)
    async def edit_list_view(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str
    ) -> Template:
        """Get edit list view."""
        records = await conn.fetch(SELECT_LIST_WITH_CIDR_COUNT, request.user.id, id)
        if not records:
            raise NotFoundException(f"List {id} not found.")
        item = dict(records[0])
        item["tags"] = ",".join(item["tags"])
        return Template(template_name="partials/list-edit.html.j2", context={"list_id": id, "item": item})

    @delete("/{id:str}", include_in_schema=False, status_code=HTTP_200_OK)
    async def delete_list(self, request: Request[User, Token, State], conn: PoolConnectionProxy, id: str) -> Response:
        """Delete list."""
        async with conn.transaction():
            record = await conn.fetch(
                "delete from list where user_id = $1 and id = $2 returning id", request.user.id, id
            )
            if not record:
                raise NotFoundException(f"List {id} not found.")
        return Response("", status_code=HTTP_200_OK)


class WebPartCidrController(Controller):
    tags = ["Web"]
    path = "/parts/cidr"

    @get("/", include_in_schema=False)
    async def cidrs(
        self,
    ) -> Template:
        """CIDRs add/delete view."""
        return Template(template_name="partials/cidrs/cidrs.html.j2")

    @post("/", include_in_schema=False)
    async def parse_raw_cidrs(
        self,
        request: Request[User, Token, State],
        conn: PoolConnectionProxy,
        data: Annotated[dict, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Template:
        """Parse raw CIDRs."""
        ipv4_valid = set()
        ipv6_valid = set()

        ipv4_valid, ipv6_valid = parse_raw_cidrs_input_as_str(raw_data=data["cidrs-raw"])

        ipv4_string = "\n".join(ipv4_valid)
        ipv6_string = "\n".join(ipv6_valid)
        total_ipv4 = len(ipv4_valid)
        total_ipv6 = len(ipv6_valid)

        lists = await conn.fetch("select id, list_type from list where user_id = $1", request.user.id)

        return Template(
            template_name="partials/cidrs/cidrs-parsed.html.j2",
            headers={"HX-Trigger-After-Swap": "scrollToBottom"},
            context={
                "ipv4_cidrs": ipv4_string,
                "ipv6_cidrs": ipv6_string,
                "total_ipv4": total_ipv4,
                "total_ipv6": total_ipv6,
                "lists": lists,
            },
        )

    @put("/", include_in_schema=False)
    async def submit_parsed_cidrs(
        self,
        request: Request[User, Token, State],
        conn: PoolConnectionProxy,
        data: Annotated[dict, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Template:
        """Submit parsed CIDRs as a Job."""
        ipv4_valid, ipv6_valid = parse_raw_cidrs_input(raw_data=data["ipv4-cidrs"] + " " + data["ipv6-cidrs"])

        cidrs = [x.compressed for x in ipaddress.collapse_addresses(iter(ipv4_valid))] + [
            x.compressed for x in ipaddress.collapse_addresses(iter(ipv6_valid))
        ]

        list_record = await conn.fetchrow(
            "select * from list where id = $1 and user_id = $2", data["list_id"], request.user.id
        )
        if not list_record:
            raise NotFoundException(f"List {id} not found.")

        cidr_job = CidrJob(
            action=ActionEnum.DELETE if data["action"] == "delete" else ActionEnum.ADD,
            list_id=list_record["id"],
            list_type=list_record["list_type"],
            list_enabled=list_record["enabled"],
            user_id=request.user.id,
            cidrs=cidrs,
            ttl=None if int(data["ttl"]) == 0 else int(data["ttl"]),
        )

        await insert_cidr_job(job=cidr_job, conn=conn)

        return Template(
            template_name="partials/cidrs/cidrs-job.html.j2",
            headers={"HX-Trigger-After-Swap": "scrollToBottom"},
            context={"job": cidr_job},
        )

    @get("/ni", include_in_schema=False)
    async def get_ni(self, request: Request[User, Token, State], conn: PoolConnectionProxy) -> Template:
        """Get cidr."""
        return await get_network_info(conn=conn, address=None, user_id=request.user.id)

    @get("/ni/{ip:str}/{prefix:str}", include_in_schema=False)
    async def get_cidr(
        self, request: Request[User, Token, State], conn: PoolConnectionProxy, ip: str, prefix: str
    ) -> Template:
        """Get cidr."""
        address = ip + "/" + prefix
        return await get_network_info(conn=conn, address=address, user_id=request.user.id)

    @post("/ni", include_in_schema=False)
    async def get_ni_cidr(
        self,
        request: Request[User, Token, State],
        conn: PoolConnectionProxy,
        data: Annotated[dict, Body(media_type=RequestEncodingType.URL_ENCODED)],
    ) -> Template:
        """Get network info for CIDR."""
        return await get_network_info(conn=conn, address=data.get("address", ""), user_id=request.user.id)

    @delete("/{list_id:str}/{ip:str}/{prefix:str}", include_in_schema=False, status_code=HTTP_200_OK)
    async def delete_cidr(
        self,
        conn: PoolConnectionProxy,
        list_id: str,
        ip: str,
        prefix: str,
    ) -> Response:
        """Delete CIDR."""
        address = ip + "/" + prefix
        async with conn.transaction():
            record = await conn.fetch(
                "delete from cidr where list_id = $1 and address = $2 returning address", list_id, address
            )
            if not record:
                raise NotFoundException(f"CIDR {address} not found.")
        return Response("", status_code=HTTP_200_OK)

src/app/domain/web/controllers.py
aorith/cidr-listings
from hashlib import blake2b
from os import urandom

from argon2 import PasswordHasher
from argon2.exceptions import VerificationError, VerifyMismatchError

ph = PasswordHasher()


async def generate_salt_and_hashed_password(plain_password: str) -> tuple[str, str]:
    """Generates a new salt and a hashed_password.

    A new salt is always generated when calling this function, that
    salt is used to generate the hash of the provided plain_password.

    :param plain_password: Plain password
    :returns: (salt, hashed_password)
    """
    # generate the salt
    b2 = blake2b(urandom(60))
    salt = b2.hexdigest()

    # append salt to the plain password and generate the final hash
    hash = ph.hash(plain_password + salt)
    return salt, hash


async def verify_password(salt: str, hashed_password: str, plain_password: str) -> bool:
    """Verifies the plain password against the hashed one."""
    try:
        return ph.verify(hashed_password, plain_password + salt)
    except VerifyMismatchError:
        # Secret does not match the password
        return False
    except VerificationError:
        # Verification failed
        raise
    except Exception as err:
        print(err)
        raise

src/app/lib/authcrypt.py
aorith/cidr-listings
import asyncio

import click
from click import Group
from litestar import Litestar
from litestar.plugins import CLIPluginProtocol

from app.domain.auth.services import create_user as create_user_service
from app.domain.auth.services import reset_password as reset_password_service
from app.lib.db.migrations import run_migrations_from_cli


def _run_task(coro) -> None:
    loop = asyncio.new_event_loop()
    task = loop.create_task(coro)  # noqa: F841
    pending = asyncio.all_tasks(loop=loop)
    group = asyncio.gather(*pending)
    loop.run_until_complete(group)
    loop.close()


class CLIPlugin(CLIPluginProtocol):
    def on_cli_init(self, cli: Group) -> None:
        @cli.command()
        def is_debug_mode(app: Litestar):
            if app.debug:
                click.echo("App IS running in Debug mode.")
            else:
                click.echo("App is NOT running in Debug mode.")

        @cli.command(name="run-db-migrations", help="Run DB Migrations")
        def run_db_migrations() -> None:
            click.echo("Running DB migrations...")
            _run_task(run_migrations_from_cli())

        @cli.command(name="create-user", help="Create a user")
        @click.option(
            "--login",
            help="Login name of the new user",
            type=click.STRING,
            required=True,
            show_default=False,
        )
        @click.option(
            "--password",
            help="Password",
            type=click.STRING,
            required=True,
            show_default=False,
        )
        @click.option(
            "--superuser",
            help="Make it a superuser",
            type=click.BOOL,
            default=False,
            required=False,
            show_default=False,
            is_flag=True,
        )
        def create_user(login: str, password: str, superuser: bool = False) -> None:
            """Create a user."""
            click.echo("Running create user")
            _run_task(create_user_service(login=login, password=password, superuser=superuser))

        @cli.command(name="reset-password", help="Reset user password")
        @click.option(
            "--login",
            help="Login name of the user",
            type=click.STRING,
            required=True,
            show_default=False,
        )
        @click.option(
            "--password",
            help="New password",
            type=click.STRING,
            required=True,
            show_default=False,
        )
        def reset_password(login: str, password: str) -> None:
            """Create a user."""
            click.echo("Running reset-user")
            _run_task(reset_password_service(login=login, password=password))

src/app/lib/cli.py
aorith/cidr-listings
import asyncio
from collections.abc import AsyncGenerator
from functools import lru_cache

import asyncpg

from app.lib.settings import get_settings

settings = get_settings()

dsn = "postgres://{}:{}@{}:{}/{}".format(
    settings.DB_USERNAME,
    settings.DB_PASSWORD,
    settings.DB_HOST,
    settings.DB_PORT,
    settings.DB_NAME,
)


async def get_connection() -> AsyncGenerator[asyncpg.Connection, None]:
    """Get a connection to the DB."""
    conn = await asyncpg.connect(dsn=dsn)
    try:
        yield conn
    finally:
        await conn.close()


class DBManager:
    pool: asyncpg.Pool

    async def setup(self) -> None:
        """Initialize the DB pool."""
        pool = await asyncpg.create_pool(
            dsn=dsn,
            timeout=30,
            min_size=settings.DB_POOL_MIN_SIZE,
            max_size=settings.DB_POOL_MAX_SIZE,
            max_inactive_connection_lifetime=settings.DB_POOL_MAX_IDLE_TIMEOUT,
        )
        if not isinstance(pool, asyncpg.Pool):
            raise Exception("Pool is not initialized")
        self.pool = pool

    async def get_connection(self) -> AsyncGenerator[asyncpg.pool.PoolConnectionProxy, None]:
        """Get a connection from the pool."""
        conn = await self.pool.acquire(timeout=settings.DB_POOL_ADQUIRE_CONN_TIMEOUT)
        try:
            yield conn
        finally:
            await self.pool.release(conn)

    async def stop(self) -> None:
        """Close the pool."""
        try:
            if hasattr(self, "pool") and isinstance(self.pool, asyncpg.Pool) and not self.pool.is_closing():
                await asyncio.wait_for(self.pool.close(), timeout=settings.DB_POOL_CLOSE_TIMEOUT)
        except asyncio.TimeoutError:
            pass


@lru_cache
def get_dbmanager() -> DBManager:
    """Get DBManager singleton."""
    return DBManager()

src/app/lib/db/base.py
aorith/cidr-listings
import asyncio
from datetime import datetime, timezone
from pathlib import Path

from asyncpg import Connection

from app.lib.db.base import get_connection

MIGRATIONS_DIR = Path(__file__).parent.resolve() / "migrations"

CHECK_MIGRATIONS_TABLE = """
SELECT EXISTS (
    SELECT FROM
        pg_tables
    WHERE
        schemaname = 'public' AND
        tablename  = '_migrations'
    );
"""

MIGRATIONS_TABLE_DDL = """
BEGIN;
CREATE TABLE _migrations (
    version INTEGER NOT NULL PRIMARY KEY,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
INSERT INTO _migrations (version) VALUES (-1);
COMMIT;
"""


async def create_migrations_table(conn: Connection) -> None:
    """Create migrations table."""
    print(">>> Creating migrations table ...")
    await conn.execute(MIGRATIONS_TABLE_DDL)


async def get_current_version(conn: Connection) -> int | None:
    """Get current DB version."""
    print("- Retrieving current version ...")
    ver = await conn.fetchval("select version from _migrations")
    return int(ver) if ver else None


async def run_migration(conn: Connection, migrations_dir: Path, sql: str, sql_ver: int) -> None:
    """Run migration."""
    print(f">>> Executing SQL migration = {sql}")
    with (migrations_dir / Path(sql)).open("r") as fd:
        data = fd.read().encode()
    await conn.execute(data.decode())
    await conn.execute("update _migrations set version = $1, updated_at = $2", sql_ver, datetime.now(tz=timezone.utc))
    print(f"- Current version is now = {sql_ver}")


async def run_migrations() -> int:
    """Run migrations."""
    async for conn in get_connection():
        async with conn.transaction():
            if await conn.fetchval(CHECK_MIGRATIONS_TABLE):  # returns True or False directly
                curr_ver = await get_current_version(conn)
            else:
                await create_migrations_table(conn)
                curr_ver = -1

            print(f"- Current version = {curr_ver}")
            if curr_ver is None:
                raise ValueError("Current version cannot be None, _migrations table wasn't created successfully.")
            print("- Reading migration files ...")
            migration_sqls = sorted([x.name for x in MIGRATIONS_DIR.glob("*.sql")])
            for sql in migration_sqls:
                sql_ver = int(sql.split("_")[0])
                if sql_ver > curr_ver:
                    await run_migration(conn, MIGRATIONS_DIR, sql, sql_ver)
                print(f"- {sql}  ... OK")
    return 0


async def run_migrations_from_cli() -> None:
    """Run migrations from cli."""
    await run_migrations()


async def main() -> int:
    """Run migrations."""
    return await run_migrations()


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

src/app/lib/db/migrations.py
aorith/cidr-listings
from datetime import datetime, timezone


def datetime_no_microseconds() -> datetime:
    return datetime.now(tz=timezone.utc).replace(microsecond=0)

src/app/lib/default_factories.py
aorith/cidr-listings
from litestar import MediaType, Request, Response
from litestar.datastructures import Cookie
from litestar.response import Redirect
from litestar.status_codes import (
    HTTP_204_NO_CONTENT,
    HTTP_307_TEMPORARY_REDIRECT,
    HTTP_401_UNAUTHORIZED,
    HTTP_500_INTERNAL_SERVER_ERROR,
)

from app.lib.settings import get_settings

settings = get_settings()


def default_httpexception_handler(request: Request, exc: Exception) -> Response:
    """Handler for all exceptions subclassed from HTTPException."""  # noqa: D401
    status_code = getattr(exc, "status_code", HTTP_500_INTERNAL_SERVER_ERROR)

    if request.url.path.startswith("/v1/"):
        preferred_type = MediaType.JSON
        extra = getattr(exc, "extra", "")
        content = {"status_code": status_code, "detail": getattr(exc, "detail", "")}
        if extra:
            content.update({"extra": extra})
    else:
        if not request.url.path.startswith("/login") and status_code == HTTP_401_UNAUTHORIZED:
            # redirect to login and reset the cookie
            if request.url.path.startswith("/parts"):
                return Response(
                    None,
                    status_code=HTTP_204_NO_CONTENT,  # htmx redirects need a 2xx
                    headers={"HX-Redirect": "/login"},
                    media_type=MediaType.HTML,
                    cookies=[Cookie(key=settings.API_KEY_COOKIE, value="", samesite="strict", max_age=-1)],
                )
            return Redirect(
                "/login",
                status_code=HTTP_307_TEMPORARY_REDIRECT,
                media_type=MediaType.HTML,
                cookies=[Cookie(key=settings.API_KEY_COOKIE, value="", samesite="strict", max_age=-1)],
            )

        preferred_type = MediaType.HTML
        detail = getattr(exc, "detail", "")  # litestar exceptions
        extras = getattr(exc, "extra", "")  # msgspec exceptions
        if extras and len(extras) >= 1:
            message = extras[0]
            if isinstance(message, dict):
                detail = f"{message.get('key', '')}: {message.get('message', detail)}"

        content = f"<p><span class='error'>{detail}</span></p>"

    return Response(
        media_type=preferred_type,
        content=content,
        headers={"HX-Trigger": "cleanErrors"},
        status_code=status_code,
    )

src/app/lib/exceptions.py
aorith/cidr-listings
from collections.abc import Generator
from ipaddress import (
    IPv4Address,
    IPv4Network,
    IPv6Address,
    IPv6Network,
    collapse_addresses,
    summarize_address_range,
)


def exclude_address_raw(
    base_range: tuple[int, int],
    exclude_range: tuple[int, int],
) -> Generator[tuple[int, int], None, None]:
    """Excludes a network from another.

    The range can be obtained from an ipaddress.IPv[4|6]Network object by casting it
    to an integer or accessing the protected field ``_ip``, i.e.:

        ```
        ipn = IPv4Network("10.10.0.0/24")
        ipn.network_address._ip, ipn.broadcast_address._ip
        (168427520, 168427775)
        ```

    Both ranges must be of the same IP version.

    Args:
    ----
        base_range (tuple[int, int]): IP Range to be excluded
        exclude_range(tuple[int, int]): IP Range that tries to exclude base_range

    Returns:
    -------
        Generator[tuple[int, int], None, None]: IP Ranges left after the exclusion
    """
    # --x--x-|--|-----
    # -------|--|-x-x-
    if base_range[0] > exclude_range[1] or base_range[1] < exclude_range[0]:
        # exclude_range does not overlap with base_range, we return the full base_range
        yield base_range
    # ---x-|------|-x-
    # -----X------|-x-
    # ---x-|------X---
    # -----X------X---
    elif base_range[0] >= exclude_range[0] and base_range[1] <= exclude_range[1]:
        # exclude_range fully overlaps base_range, we return nothing
        return
    # ---x--|---x--|--
    elif base_range[0] > exclude_range[0] and base_range[1] > exclude_range[1]:
        yield exclude_range[1] + 1, base_range[1]
    # ---|--x---|--x--
    elif base_range[0] < exclude_range[0] and base_range[1] < exclude_range[1]:
        yield base_range[0], exclude_range[0] - 1
    # -----X---x--|---
    elif base_range[0] == exclude_range[0]:
        # exclude range starts together with base_range and ends before base_range
        yield exclude_range[1] + 1, base_range[1]
    # -----|--x---X---
    elif base_range[1] == exclude_range[1]:
        # exclude range ends together with base_range and starts after base_range
        yield base_range[0], exclude_range[0] - 1
    # ---|--x---x--|--
    else:
        # exclude range creates two new ranges
        yield base_range[0], exclude_range[0] - 1
        yield exclude_range[1] + 1, base_range[1]


def exclude_address(
    base_net: IPv4Network,
    exclude_net: IPv6Network,
    base_class,
) -> Generator[IPv4Network | IPv6Network, None, None]:
    """Excludes a network from another.

    Accepts and returns ipaddress network objects.
    """
    base_range = base_net.network_address._ip, base_net.broadcast_address._ip  # type: ignore
    exclude_range = exclude_net.network_address._ip, exclude_net.broadcast_address._ip  # type: ignore

    for subn_range in exclude_address_raw(base_range=base_range, exclude_range=exclude_range):
        yield from summarize_address_range(base_class(subn_range[0]), base_class(subn_range[1]))


async def address_exclude_many(
    cidr: IPv4Network | IPv6Network, exclusion_cidrs: set[IPv4Network | IPv6Network]
) -> set[IPv4Network | IPv6Network]:
    """Excludes addresses from ``cidr`` present in any ``exclusion_cidrs``, subnetting if necessary."""

    async def _exclude_from_subnets(_subnet_ranges, _exclusion):
        exclude_range = _exclusion.network_address._ip, _exclusion.broadcast_address._ip
        new_subnet_ranges = set()
        for isub_range in _subnet_ranges:
            for _subnet_range in exclude_address_raw(base_range=isub_range, exclude_range=exclude_range):
                new_subnet_ranges.add(_subnet_range)
        return new_subnet_ranges

    base_class = IPv4Address if cidr.version == 4 else IPv6Address

    subnet_ranges = {(cidr.network_address._ip, cidr.broadcast_address._ip)}  # type:ignore
    for exclusion_cidr in (x for x in exclusion_cidrs if x.version == cidr.version):
        subnet_ranges = await _exclude_from_subnets(subnet_ranges, exclusion_cidr)
        if not subnet_ranges:
            break

    final_subnets = set()
    for subnet_range in subnet_ranges:
        for subn in summarize_address_range(base_class(subnet_range[0]), base_class(subnet_range[1])):
            final_subnets.add(subn)

    return set(collapse_addresses(iter(final_subnets)))  # type: ignore


async def address_exclude_many_2(
    cidr: IPv4Network | IPv6Network, exclusion_cidrs: set[IPv4Network | IPv6Network]
) -> set[IPv4Network | IPv6Network]:
    """Excludes addresses from ``cidr`` present in any ``exclusion_cidrs``, subnetting if necessary.

    Slower than address_exclude_many
    """

    def _exclude_from_subnets(_subnets, _exclusion):
        new_subnets = set()
        for isub in _subnets:
            if not isub.subnet_of(_exclusion):
                try:
                    for isub2 in isub.address_exclude(_exclusion):
                        new_subnets.add(isub2)
                except ValueError:
                    new_subnets.add(isub)
                    continue
        return new_subnets

    subnets = {cidr}
    for exclusion_cidr in (x for x in exclusion_cidrs if x.version == cidr.version):
        if not subnets:
            break
        subnets = _exclude_from_subnets(subnets, exclusion_cidr)
    return set(collapse_addresses(iter(subnets)))  # type: ignore

src/app/lib/iputils.py
aorith/cidr-listings
from litestar.openapi.config import OpenAPIConfig
from litestar.openapi.spec import Components, Contact, License, SecurityScheme

from app.lib.settings import get_settings

settings = get_settings()


openapi_config = OpenAPIConfig(
    title=settings.OPENAPI_TITLE,
    version=settings.VERSION,
    contact=Contact(name=settings.OPENAPI_CONTACT_NAME, email=settings.OPENAPI_CONTACT_EMAIL),
    license=License(name="MIT", identifier="MIT"),
    use_handler_docstrings=True,
    root_schema_site="swagger",
    path=settings.OPENAPI_PATH,
    components=Components(
        security_schemes={
            "BearerToken": SecurityScheme(
                type="http",
                scheme="bearer",
                bearer_format="JWT",
                security_scheme_in="header",
                description=(
                    "API Key Header in the format `Bearer <TOKEN>`, "
                    "you must include the `Bearer` string along with the token"
                ),
            ),
            # "CookieToken": SecurityScheme( # this doesn't actually set the cookie in swagger
            # ),
        },
    ),
)

src/app/lib/openapi.py
aorith/cidr-listings
import asyncio
import threading
from abc import ABC, abstractmethod

from app.lib.db.base import get_connection
from app.lib.settings import get_settings

settings = get_settings()


class ScheduledTask(ABC):
    keep_running: bool

    def __init__(self) -> None:
        self.keep_running = True
        self.name = self.__class__.__name__

    def stop(self) -> None:
        print(f"Stopping {self.name} ...")
        self.keep_running = False

    async def run(self) -> None:
        t = threading.Thread(target=asyncio.run, args=(self._execute_loop(),), name=self.name, daemon=True)
        t.start()
        print(f"Starting Scheduler {self.name} - {self.keep_running=}")

    async def run_once(self) -> None:
        """Run the scheduled task once and quit.

        Mostly used for testing.
        """
        await self._execute()

    @abstractmethod
    async def _execute_loop(self) -> None:
        pass

    @abstractmethod
    async def _execute(self) -> None:
        pass


class TaskDeleteExpired(ScheduledTask):
    async def _execute_loop(self) -> None:
        try:
            while self.keep_running:
                await self._execute()
                await asyncio.sleep(settings.SCHEDULER_DELETE_EXPIRED_INTERVAL)
        except KeyboardInterrupt:
            self.keep_running = False

    async def _execute(self) -> None:
        """Delete expired CIDRs."""
        async for conn in get_connection():
            async with conn.transaction():
                res = await conn.execute("delete from cidr where expires_at < now()")
                print(f"Delete expired CIDRs task: {res}")


class Scheduler:
    """Scheduled tasks."""

    tasks: list[ScheduledTask]

    def __init__(self):
        self.tasks = [TaskDeleteExpired()]

    def stop(self):
        print("Stopping Scheduler.")
        for task in self.tasks:
            task.stop()

    async def run(self) -> None:
        """Start scheduled threads."""
        for task in self.tasks:
            await task.run()

src/app/lib/scheduled_tasks.py
aorith/cidr-listings
from functools import lru_cache

from pydantic import Field, ValidationError
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    """Pydantic class to manage environment variables.

    Default values are overriden by the environment
    variables with the same name but in uppercase.
    """

    # DB
    DB_USERNAME: str = "test"
    DB_PASSWORD: str = "test1234"
    DB_HOST: str = "127.0.0.1"
    DB_PORT: int = 5432
    DB_NAME: str = "cidr"
    DB_POOL_MIN_SIZE: int = 5
    """Minimal number of connections that a pool starts with."""
    DB_POOL_MAX_SIZE: int = 10
    """Maximum number of connections that a pool can allocate."""
    DB_POOL_MAX_IDLE_TIMEOUT: int = Field(default=300, ge=30)
    """Maximum time a connection can be idle."""
    DB_POOL_ADQUIRE_CONN_TIMEOUT: int = 5
    """Maximum time allowed to get a connection from the pool."""
    DB_POOL_CLOSE_TIMEOUT: int = 10
    """Timeout to close the pool."""

    # Worker
    JOB_QUEUE_QUERY_INTERVAL: int = 5
    """Interval between the DB query that fetches the jobs from the 'job_queue' table."""

    # Scheduler tasks
    SCHEDULER_DELETE_EXPIRED_INTERVAL: int = 30
    """Interval between the task that deletes expired CIDRs."""

    # APP
    VERSION: str = "1.0"
    DEBUG: bool = False

    # JWT
    JWT_SECRET: str
    ALGORITHM: str = "HS256"
    API_KEY_COOKIE: str = "apisessionkey"

    DEFAULT_TOKEN_TTL_SECONDS: int = 60 * 60 * 24
    """Expiration time for the token"""

    AUTH_CACHE_SECONDS: int = 120
    """Internal TTL for the cached authentication results"""

    # OpenAPI
    OPENAPI_TITLE: str = "CIDR Listings"
    OPENAPI_CONTACT_NAME: str = "Manuel Sanchez Pinar"
    OPENAPI_CONTACT_EMAIL: str = "aomanu@gmail.com"
    OPENAPI_PATH: str = "/docs"


@lru_cache
def get_settings():
    """Return cached settings so it's the same instance everywhere."""
    try:
        return Settings()  # type: ignore
    except ValidationError as err:
        print("Could not load settings.", err)  # noqa: T201
        raise

src/app/lib/settings.py
aorith/cidr-listings
from typing import Any

import msgspec
from litestar.exceptions import ValidationException
from msgspec import ValidationError


async def run_validation(data: Any, target_type: Any):
    """Run the msgspec validation wrapped in a try/except.

    Validations aren't run automatically by msgpec when
    wrapped into a DTO.

    ValidationError's aren't handled in routes automatically
    when DTOs are involved.
    """
    # Until something like this is implemented: https://github.com/jcrist/msgspec/issues/513
    # but this is better for performance reasons, we only validate incoming data
    try:
        msgspec.msgpack.decode(msgspec.msgpack.encode(data), type=target_type)
    except ValidationError as err:
        raise ValidationException(str(err))  # noqa: B904

src/app/lib/validations.py
aorith/cidr-listings
import asyncio
import ipaddress
import threading
import time
from collections import Counter
from datetime import datetime, timedelta, timezone
from ipaddress import IPv4Network, IPv6Network, collapse_addresses, ip_network
from uuid import UUID

import msgspec
from asyncpg import Connection

from app.domain.lists.schemas import ActionEnum, CidrJob, ListTypeEnum
from app.lib.db.base import get_connection
from app.lib.iputils import address_exclude_many
from app.lib.settings import get_settings

settings = get_settings()

cidrjob_dec = msgspec.json.Decoder(type=CidrJob)

CONSUME_JOB_QUERY = """
DELETE FROM
    job_queue
USING (
    SELECT * FROM job_queue ORDER BY id FOR UPDATE SKIP LOCKED
) q
WHERE q.id = job_queue.id RETURNING job_queue.*;
"""

UPSERT_CIDR = """
INSERT INTO cidr
    (address, list_id, expires_at)
VALUES
    ($1, $2, $3)
ON CONFLICT (address, list_id)
DO
    UPDATE SET expires_at = $3
"""


SELECT_ENABLED_CIDRS_BY_LIST_TYPE = """
SELECT
    address, list_id, expires_at
FROM
    cidr
WHERE
    list_id IN (
            SELECT id FROM list WHERE
                enabled = true
            AND
                user_id = $1
            AND
                list_type = $2
        )
"""

SELECT_ENABLED_CIDRS_BY_LIST_ID = """
SELECT
    address, list_id, expires_at
FROM
    cidr
WHERE
    list_id = (
            SELECT id FROM list WHERE
                enabled = true
            AND
                id = $1
        )
"""

SELECT_ALL_CIDRS_BY_LIST_ID = """
SELECT
    address, list_id, expires_at
FROM
    cidr
WHERE
    list_id = (
            SELECT id FROM list WHERE
                id = $1
        )
"""


async def parse_raw_cidrs(
    cidrs: list[str], only_global: bool = True
) -> tuple[Counter, set[IPv4Network], set[IPv6Network]]:
    """Parse the initial list of CIDRs coming from a job.

    1. Converts the str representation to an IPv4Network/IPv6Network object, filtering malformed
    2. Filters non routable/non global addresses if ``only_global`` is True
    3. Splits IPv4Network and IPv6Network is two sets
    4. Collapses the two sets
    5. Returns a counter with the parsing information a set of IPv4Networks and another of IPv6Networks
    """
    result = Counter(total_job=0, malformed=0, non_global=0)
    ipv4_cidrs: set[IPv4Network] = set()
    ipv6_cidrs: set[IPv6Network] = set()

    for cidr in cidrs:
        result["total_job"] += 1
        try:
            ipn = ipaddress.ip_network(cidr)
            if not ipn.is_global and only_global:
                result["not_global"] += 1
                continue
        except ValueError:
            result["malformed"] += 1
            continue

        if ipn.version == 4:
            ipv4_cidrs.add(ipn)
        elif ipn.version == 6:
            ipv6_cidrs.add(ipn)
        else:
            raise ValueError(f"Unknown IP Version {ipn.version}")

    ipv4_cidrs = set(collapse_addresses(iter(ipv4_cidrs)))
    ipv6_cidrs = set(collapse_addresses(iter(ipv6_cidrs)))

    return result, ipv4_cidrs, ipv6_cidrs


async def filter_safe_cidrs(
    conn: Connection,
    user_id: UUID,
    cidrs: set[IPv4Network | IPv6Network],
) -> tuple[set[IPv4Network], set[IPv6Network]]:
    """Filter input ``cidrs`` that are present on enabled lists of type SAFE for ``user_id``."""
    safe_cidrs = set()
    for safe_cidr_record in await conn.fetch(
        "select address from cidr where list_id in (select id from list where enabled = true and list_type = $1 and user_id = $2)",
        ListTypeEnum.SAFE,
        user_id,
    ):
        safe_cidrs.add(safe_cidr_record["address"])

    safe_cidr_ipv4 = {x for x in safe_cidrs if x.version == 4}
    safe_cidr_ipv6 = {x for x in safe_cidrs if x.version == 6}

    deny_subnets = set()
    for deny_cidr in cidrs:
        for subnet in await address_exclude_many(
            cidr=deny_cidr,
            exclusion_cidrs=safe_cidr_ipv4 if deny_cidr.version == 4 else safe_cidr_ipv6,  # type: ignore
        ):
            deny_subnets.add(subnet)

    return (
        set(collapse_addresses(x for x in deny_subnets if x.version == 4)),
        set(collapse_addresses(x for x in deny_subnets if x.version == 6)),
    )


async def delete_excluded_cidrs(
    conn: Connection,
    user_id: UUID,
    exclusion_cidrs: set[IPv4Network | IPv6Network],
    list_id: str | None = None,
    list_type: ListTypeEnum | None = None,
) -> None:
    """Delete CIDRs matching ``exclusion_cidrs``.

    This function cleans the DB from the excluded addresses preserving
    the ones not excluded instead of deleting the whole subnet.

    Use case example:
    When adding new CIDRs to a safelist we need to clean existing
    denylists to honor the new additions to the safelist.
    If a CIDR from the safelist is a subnet of a CIDR from a
    denylist, the CIDR from the denylist will be split into
    subnets that do not contain the one in the safelist.
    """
    if list_id:
        exclusion_record = await conn.fetch(SELECT_ALL_CIDRS_BY_LIST_ID, list_id)
    elif list_type:
        exclusion_record = await conn.fetch(SELECT_ENABLED_CIDRS_BY_LIST_TYPE, user_id, list_type)
    else:
        raise NotImplementedError("Either `list_id` or `list_type` is needed.")
    if not exclusion_record:
        return

    exclusion_cidr_ipv4 = {x for x in exclusion_cidrs if x.version == 4}
    exclusion_cidr_ipv6 = {x for x in exclusion_cidrs if x.version == 6}
    new_subnets = []
    to_delete = set()

    for record in exclusion_record:
        exclusion_cidr = ipaddress.ip_network(str(record["address"]))
        exclusion_subnets = await address_exclude_many(
            cidr=exclusion_cidr,
            exclusion_cidrs=exclusion_cidr_ipv4 if exclusion_cidr.version == 4 else exclusion_cidr_ipv6,  # type: ignore
        )

        if len(exclusion_subnets) == 1:
            subnet = exclusion_subnets.pop()
            if subnet == exclusion_cidr:
                # the same address has been returned so no filtering was needed, we
                # upsert to update expires_at
                new_subnets.append((subnet, record["list_id"], record["expires_at"]))
            else:
                # one subnet left which isn't the original, upsert the new and delete the old
                to_delete.add((exclusion_cidr, record["list_id"]))
                new_subnets.append((subnet, record["list_id"], record["expires_at"]))
        else:
            # either no subnets returned or many which means the original needs to be deleted
            to_delete.add((exclusion_cidr, record["list_id"]))
            for exc_subnet in exclusion_subnets:
                new_subnets.append((exc_subnet, record["list_id"], record["expires_at"]))

    # Execute the queries
    await conn.executemany("delete from cidr where address = $1 and list_id = $2", list(to_delete))
    await conn.executemany(UPSERT_CIDR, new_subnets)


async def add_cidrs(conn: Connection, cidr_job: CidrJob) -> None:
    """Add the CIDRs included in the job."""
    stime = time.perf_counter()

    # Initial parsing
    result, ipv4_cidrs_parsed, ipv6_cidrs_parsed = await parse_raw_cidrs(cidrs=cidr_job.cidrs)

    # Remove special addresses that should not touch the DB, like 0.0.0.0
    ipv4_cidrs: set[IPv4Network] = set()
    for c in ipv4_cidrs_parsed:
        if c.prefixlen == 0:
            continue
        if c.compressed.split("/")[0] == "0.0.0.0":  # noqa: S104
            continue
        ipv4_cidrs.add(c)
    ipv4_cidrs_parsed.clear()

    ipv6_cidrs: set[IPv6Network] = set()
    for c in ipv6_cidrs_parsed:
        if c.prefixlen == 0:
            continue
        ipv6_cidrs.add(c)
    ipv6_cidrs_parsed.clear()

    if not ipv4_cidrs and not ipv6_cidrs:
        print(f"Add({cidr_job.list_type}): {result}")
        return

    if cidr_job.list_type == ListTypeEnum.DENY:
        # When adding CIDRs to a deny list, we only need to filter out CIDRs in current safe lists
        ipv4_cidrs, ipv6_cidrs = await filter_safe_cidrs(
            conn=conn, user_id=cidr_job.user_id, cidrs=ipv4_cidrs | ipv6_cidrs
        )
    else:
        # When adding CIDRs to a safe list, we must delete matching CIDRs in current deny lists
        # if the target safe list is enabled
        if cidr_job.list_enabled:
            await delete_excluded_cidrs(
                conn=conn,
                user_id=cidr_job.user_id,
                exclusion_cidrs=ipv4_cidrs | ipv6_cidrs,
                list_type=ListTypeEnum.DENY,
            )

    sql_params = []
    for cidr in ipv4_cidrs | ipv6_cidrs:
        result["total_final"] += 1
        expires_at = datetime.now(tz=timezone.utc) + timedelta(seconds=cidr_job.ttl) if cidr_job.ttl else None
        sql_params.append((cidr, cidr_job.list_id, expires_at))

    await conn.executemany(UPSERT_CIDR, sql_params)
    print(f"Add({cidr_job.list_type}): {result} - took {time.perf_counter() - stime} seconds")


async def delete_cidrs(conn: Connection, cidr_job: CidrJob) -> None:
    """Delete the CIDRs included in the job."""
    stime = time.perf_counter()

    # Initial parsing
    result, ipv4_cidrs, ipv6_cidrs = await parse_raw_cidrs(cidrs=cidr_job.cidrs, only_global=False)
    if not ipv4_cidrs and not ipv6_cidrs:
        return

    await delete_excluded_cidrs(
        conn=conn, user_id=cidr_job.user_id, exclusion_cidrs=ipv4_cidrs | ipv6_cidrs, list_id=cidr_job.list_id
    )

    print(f"Delete({cidr_job.list_type}): {result} - took {time.perf_counter() - stime} seconds")


async def update_cleanup(conn: Connection, cidr_job: CidrJob) -> None:
    """Do a CIDR cleanup from denylists when a safelist is re-enabled."""
    stime = time.perf_counter()

    if cidr_job.list_type != ListTypeEnum.SAFE:
        raise ValueError("update_cleanup() should only process safe lists.")

    # This job doesn't actually carry the CIDRs from the safelist, we get them here
    num_addresses = 0
    all_addresses = set()
    for record in await conn.fetch(SELECT_ENABLED_CIDRS_BY_LIST_ID, cidr_job.list_id):
        num_addresses += 1
        all_addresses.add(ip_network(record["address"]))
    ipv4_collapsed = set(collapse_addresses(x for x in all_addresses if x.version == 4))
    ipv6_collapsed = set(collapse_addresses(x for x in all_addresses if x.version == 6))

    await delete_excluded_cidrs(
        conn=conn,
        user_id=cidr_job.user_id,
        exclusion_cidrs=ipv4_collapsed | ipv6_collapsed,
        list_type=ListTypeEnum.DENY,
    )
    print(f"UpdateCleanup({cidr_job.list_type}): for {num_addresses} cidrs took {time.perf_counter() - stime} seconds")


class CidrWorker:
    """Consumes the job queue that inserts and deletes CIDRs in/from lists."""

    keep_running: bool

    def __init__(self) -> None:
        self.keep_running = True

    def stop(self) -> None:
        """Stop CidrWorker consume_loop."""
        print("Stopping CidrWorker.")
        self.keep_running = False

    async def run(self) -> None:
        """Start the bg thread with consume_loop."""
        t = threading.Thread(
            target=asyncio.run, args=(self._consume_loop(),), name="job_queue_worker_thread", daemon=True
        )
        t.start()

    async def run_once(self) -> None:
        """Consume one batch of jobs and quit.

        Mostly used for testing.
        """
        await self._process_jobs()

    async def _consume_loop(self) -> None:
        """Consume jobs in a loop."""
        try:
            print(f"Starting CidrWorker.consume_loop() - {self.keep_running=}")
            while self.keep_running:
                await self._process_jobs()
                await asyncio.sleep(settings.JOB_QUEUE_QUERY_INTERVAL)
        except KeyboardInterrupt:
            self.keep_running = False

    async def _process_jobs(self) -> None:
        """Process one batch of jobs."""
        async for conn in get_connection():
            async with conn.transaction():
                for record in await conn.fetch(CONSUME_JOB_QUERY):
                    cidr_job = cidrjob_dec.decode(record["payload"])
                    if cidr_job.action == ActionEnum.ADD:
                        await add_cidrs(conn=conn, cidr_job=cidr_job)
                    elif cidr_job.action == ActionEnum.DELETE:
                        await delete_cidrs(conn=conn, cidr_job=cidr_job)
                    elif cidr_job.action == ActionEnum.UPDATE:
                        await update_cleanup(conn=conn, cidr_job=cidr_job)

src/app/lib/worker.py
aorith/cidr-listings
from pathlib import Path

from litestar import Litestar
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.datastructures import ResponseHeader
from litestar.di import Provide
from litestar.exceptions import HTTPException
from litestar.middleware.base import DefineMiddleware
from litestar.static_files.config import StaticFilesConfig
from litestar.template.config import TemplateConfig

from app.domain import routes
from app.domain.auth.middleware import JWTAuthenticationMiddleware
from app.lib.cli import CLIPlugin
from app.lib.db.base import get_dbmanager
from app.lib.db.migrations import run_migrations
from app.lib.exceptions import default_httpexception_handler
from app.lib.openapi import openapi_config
from app.lib.scheduled_tasks import Scheduler
from app.lib.settings import get_settings
from app.lib.worker import CidrWorker

settings = get_settings()
base_dir = Path(__file__).resolve().parent
statics_dir = base_dir / Path("domain/web/statics")
templates_dir = base_dir / Path("domain/web/templates")

__version__ = settings.VERSION

auth_mw = DefineMiddleware(
    JWTAuthenticationMiddleware,
    exclude=[
        "^/docs",
        "^/static/",
        "^/favicon.ico$",
        "^/login",
        "^/v1/auth",
    ],
    exclude_http_methods=["OPTIONS"],
)

dbmngr = get_dbmanager()
cidr_worker = CidrWorker()
scheduler = Scheduler()


app = Litestar(
    debug=settings.DEBUG,
    openapi_config=openapi_config,
    route_handlers=routes,
    response_headers=[ResponseHeader(name="Vary", value="Accept-Encoding", description="Default vary header")],
    exception_handlers={HTTPException: default_httpexception_handler},
    plugins=[CLIPlugin()],
    middleware=[auth_mw],
    dependencies={"conn": Provide(dbmngr.get_connection)},
    on_startup=[dbmngr.setup, run_migrations, cidr_worker.run, scheduler.run],
    on_app_init=[],
    on_shutdown=[cidr_worker.stop, scheduler.stop, dbmngr.stop],
    static_files_config=[StaticFilesConfig(directories=[statics_dir], path="/", html_mode=False)],
    template_config=TemplateConfig(directory=templates_dir, engine=JinjaTemplateEngine),
)

src/app/main.py
aorith/cidr-listings
import os
import time

import pytest
from litestar.status_codes import HTTP_200_OK
from litestar.testing import AsyncTestClient

from app.lib.settings import get_settings
from app.main import app

settings = get_settings()

TEST_USER_PAYLOAD = {
    "login": os.getenv("TEST_USER", "test"),
    "password": os.getenv("TEST_USER_PASSWORD", "Ilovet3st!"),
}

# Scopes: function > class > module > package > session


async def get_api_token_header(client):
    response = await client.post("/v1/auth/token", json=TEST_USER_PAYLOAD)
    assert response.status_code == HTTP_200_OK
    return {"Authorization": f"Bearer {response.json()['access_token']}"}


@pytest.fixture(scope="session")
def test_client() -> AsyncTestClient:
    settings.JOB_QUEUE_QUERY_INTERVAL = 1
    settings.SCHEDULER_DELETE_EXPIRED_INTERVAL = 1
    async_test_client = AsyncTestClient(app=app)
    time.sleep(1)  # wait for scheduler and worker to stop
    return async_test_client

tests/conftest.py
aorith/cidr-listings
from conftest import get_api_token_header
from litestar.status_codes import (
    HTTP_200_OK,
    HTTP_201_CREATED,
    HTTP_400_BAD_REQUEST,
    HTTP_401_UNAUTHORIZED,
    HTTP_409_CONFLICT,
)
from litestar.testing import AsyncTestClient

from app.lib.settings import get_settings

settings = get_settings()


async def test_signup(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        payload = {"login": "test01", "password": "abcdefgHij1"}
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_409_CONFLICT
        response = await client.post("/v1/admin/signup", json=payload, headers={})
        assert response.status_code == HTTP_401_UNAUTHORIZED

        response = await client.post("/v1/auth/token", json=payload)
        assert response.status_code == HTTP_200_OK
        resp_json = response.json()
        assert resp_json["token_type"] == "Bearer"
        assert resp_json["expires_in"] == settings.DEFAULT_TOKEN_TTL_SECONDS
        assert "access_token" in resp_json
        assert "expires_at" in resp_json

        response = await client.get(
            "/v1/list", headers={"Authorization": f"{resp_json['token_type']} {resp_json['access_token']}"}
        )
        assert response.status_code == HTTP_200_OK
        response = await client.get("/v1/list", headers={"Authorization": f"{resp_json['token_type']} ADFJ"})
        assert response.status_code == HTTP_401_UNAUTHORIZED
        response = await client.get(
            "/v1/list", headers={"Authorization": f"BeaRER {resp_json['access_token']}"}
        )
        assert response.status_code == HTTP_401_UNAUTHORIZED

        # Verify validations
        payload = {"login": "test0001", "password": "aaaaAF4"}
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_400_BAD_REQUEST
        payload = {"login": "test0001", "password": "aaaabbbbcccddd1"}
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_400_BAD_REQUEST

        # Create a new login for password change verification
        payload = {"login": "testc01", "password": "aaaaaaaaaAF40"}
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        # New password cannot be equal to old
        payload = {"login": "testc01", "password": "aaaaaaaaaAF40", "new_password": "aaaaaaaaaAF40"}
        response = await client.put("/v1/auth/password", json=payload)
        assert response.status_code == HTTP_400_BAD_REQUEST
        # New password validations
        payload = {"login": "testc01", "password": "aaaaaaaaaAF40", "new_password": "aaaa0"}
        response = await client.put("/v1/auth/password", json=payload)
        assert response.status_code == HTTP_400_BAD_REQUEST
        # Correct new password
        payload = {"login": "testc01", "password": "aaaaaaaaaAF40", "new_password": "newPasssswordOK01"}
        response = await client.put("/v1/auth/password", json=payload)
        assert response.status_code == HTTP_200_OK
        # Login with the old password = 401
        payload = {"login": "testc01", "password": "aaaaaaaaaAF40"}
        response = await client.post("/v1/auth/token", json=payload, follow_redirects=False)
        assert response.status_code == HTTP_401_UNAUTHORIZED

        # Successful login with the new password
        payload = {"login": "testc01", "password": "newPasssswordOK01"}
        response = await client.post("/v1/auth/token", json=payload)
        assert response.status_code == HTTP_200_OK
        payload = {"login": "testc01", "password": "newPasssswordOK01"}
        response = await client.post("/v1/auth/token", json=payload, follow_redirects=False)
        assert response.status_code == HTTP_200_OK

        # Test that only the superuser can create other users
        payload = {"login": "testc01", "password": "newPasssswordOK01"}
        response = await client.post("/v1/auth/token", json=payload)
        assert response.status_code == HTTP_200_OK
        user_token_header = {"Authorization": f"Bearer {response.json()['access_token']}"}

        payload = {"login": "newuser001", "password": "abcdefgHij1"}
        response = await client.post("/v1/admin/signup", json=payload, headers=user_token_header)
        assert response.status_code == HTTP_401_UNAUTHORIZED

tests/test_auth.py
aorith/cidr-listings
import pytest
from conftest import get_api_token_header
from litestar.status_codes import HTTP_200_OK, HTTP_201_CREATED
from litestar.testing import AsyncTestClient

from app.domain.lists.schemas import ListTypeEnum
from app.lib.worker import CidrWorker


@pytest.mark.asyncio
async def test_cidr_filters(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)
        worker = CidrWorker()

        list_safe_1 = {
            "enabled": True,
            "id": "TEST_TAGS_SAFE1",
            "list_type": ListTypeEnum.SAFE,
            "tags": ["TAG1", "COMMON"],
        }
        list_safe_2 = {
            "enabled": True,
            "id": "TEST_TAGS_SAFE2",
            "list_type": ListTypeEnum.SAFE,
            "tags": ["TAG2", "OK", "COMMON"],
        }

        # Safe list 1
        response = await client.post("/v1/list", json=list_safe_1, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_safe_1["id"]
        assert response.json()["list_type"] == list_safe_1["list_type"]

        # Safe list 2
        response = await client.post("/v1/list", json=list_safe_2, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_safe_2["id"]
        assert response.json()["list_type"] == list_safe_2["list_type"]

        # Add some CIDRs to the SAFE list 1
        safe_cidrs_payload_1 = {"cidrs": ["6.5.4.0/24", "3.2.1.0/24"]}
        response = await client.post(
            f"/v1/list/{list_safe_1['id']}/cidr/add", json=safe_cidrs_payload_1, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # Add some CIDRs to the SAFE list 2
        safe_cidrs_payload_2 = {"cidrs": ["60.50.40.0/24", "30.20.10.0/24", "33.22.11.1/32"]}
        response = await client.post(
            f"/v1/list/{list_safe_2['id']}/cidr/add", json=safe_cidrs_payload_2, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # Run worker to consume the job_queue
        await worker.run_once()

        # Query the list 1
        response = await client.get(
            f"/v1/cidr/?list_type={ListTypeEnum.SAFE}&list_id={list_safe_1['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 2

        # Query the list 2
        response = await client.get(
            f"/v1/cidr/?list_type={ListTypeEnum.SAFE}&list_id={list_safe_2['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 3

        # Filter by tags of list 1
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=TAG1",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 2
        assert sorted(response.json()) == sorted(safe_cidrs_payload_1["cidrs"])

        # Filter by tags of list 2
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=TAG2",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 3
        assert sorted(response.json()) == sorted(safe_cidrs_payload_2["cidrs"])

        # Filter by common tag 'COMMON'
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=COMMON",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 5
        assert sorted(response.json()) == sorted(safe_cidrs_payload_1["cidrs"] + safe_cidrs_payload_2["cidrs"])

        # Filter tag XXX1, no addresses should be returned
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=XXX1",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 0

        # Filter tag TAG1
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=TAG1",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 2

        # Filter tag OK
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=OK",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 3

        # Filter by common tag 'COMMON' and by-ip-version
        response = await client.get(
            f"/v1/cidr/collapsed/by-ip-version/?list_type={ListTypeEnum.SAFE}&tags=COMMON",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()["ipv4"]) == 5
        assert sorted(response.json()["ipv4"]) == sorted(safe_cidrs_payload_1["cidrs"] + safe_cidrs_payload_2["cidrs"])

        # Disable the list1
        list_safe_1_disabled = list_safe_1.copy()
        list_safe_1_disabled["enabled"] = False
        response = await client.put(
            f"/v1/list/{list_safe_1_disabled['id']}", json=list_safe_1_disabled, headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK

        # Filter by common tag 'COMMON', only the second list should return content
        response = await client.get(
            f"/v1/cidr/collapsed/?list_type={ListTypeEnum.SAFE}&tags=COMMON",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 3
        assert sorted(response.json()) == sorted(safe_cidrs_payload_2["cidrs"])

tests/test_cidr_filters.py
aorith/cidr-listings
import ipaddress

import pytest
from conftest import get_api_token_header
from litestar.status_codes import HTTP_200_OK, HTTP_201_CREATED
from litestar.testing import AsyncTestClient

from app.domain.lists.schemas import ListTypeEnum
from app.lib.worker import CidrWorker


@pytest.mark.asyncio
async def test_job_tasks(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)
        worker = CidrWorker()

        # Deny list
        list_deny = {"enabled": True, "id": "TEST_CIDRJOB_DENY1", "list_type": ListTypeEnum.DENY}
        response = await client.post("/v1/list", json=list_deny, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_deny["id"]
        assert response.json()["list_type"] == list_deny["list_type"]

        # Safe list
        list_safe = {"enabled": True, "id": "TEST_CIDRJOB_SAFE1", "list_type": ListTypeEnum.SAFE}
        response = await client.post("/v1/list", json=list_safe, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_safe["id"]
        assert response.json()["list_type"] == list_safe["list_type"]

        # Add some CIDRs to the SAFE list
        safe_cidrs_payload = {"cidrs": ["13.1.0.0/16", "13.1.1.33"]}
        response = await client.post(
            f"/v1/list/{list_safe['id']}/cidr/add", json=safe_cidrs_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # Run worker to consume the job_queue
        await worker.run_once()

        # Query the list
        response = await client.get(
            f"/v1/cidr/?list_type={ListTypeEnum.SAFE}&list_id={list_safe['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        # ["13.1.0.0/16", "13.1.1.33"] must be collapsed into 13.1.0.0/16
        assert response.json()[0]["address"] == "13.1.0.0/16"

        # Try to add a network excluded in that range...
        payload = {"cidrs": ["13.1.1.0/24"]}
        response = await client.post(f"/v1/list/{list_deny['id']}/cidr/add", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED

        # Run worker to consume the job_queue
        await worker.run_once()

        # The network should have been filtered
        response = await client.get(
            f"/v1/cidr/?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 0

        # Add non safe deny IPs
        payload = {"cidrs": ["1.2.3.4/32", "111.22.33.0/24", "4.3.2.1/32"]}
        response = await client.post(f"/v1/list/{list_deny['id']}/cidr/add", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED

        # Run worker to consume the job_queue
        await worker.run_once()

        # Deny networks should be present
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 3
        assert sorted(response.json()) == sorted(payload["cidrs"])

        # Test that exclusions happen automatically when adding IPs to enabled safelists
        safe_cidrs_payload = {"cidrs": ["111.22.0.0/16", "4.3.0.0/16"]}
        response = await client.post(
            f"/v1/list/{list_safe['id']}/cidr/add", json=safe_cidrs_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # Run worker to consume the job_queue
        await worker.run_once()

        # Safe networks should have been removed automatically from deny lists
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 1
        assert response.json() == ["1.2.3.4/32"]

        # Test deletion
        # 1. ensure the IP is on the list
        payload = {"cidrs": ["111.22.0.0/16"]}
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.SAFE}&list_id={list_safe['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert payload["cidrs"][0] in response.json()
        # 2. delete it
        response = await client.post(f"/v1/list/{list_safe['id']}/cidr/delete", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        # 3. Run worker to consume the job_queue
        await worker.run_once()
        # 4. Ensure the IP was deleted
        payload = {"cidrs": ["111.22.0.0/16"]}
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.SAFE}&list_id={list_safe['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert payload["cidrs"][0] not in response.json()

        # test that adding a IPv4Network to a denylist splits it in subnets when
        # an enabled safelists is present
        # safe: 66.66.1.0/26
        # deny: 66.66.1.0/24
        # after exclusions: 66.66.1.128/25, 66.66.1.64/26

        # 1. Delete current cidrs
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        delete_payload = {"cidrs": response.json()}
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/delete", json=delete_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # 2. Run worker to consume the job_queue
        await worker.run_once()

        # 3. Add the safe cidr
        safe_cidrs_payload = {"cidrs": ["66.66.1.0/26"]}
        response = await client.post(
            f"/v1/list/{list_safe['id']}/cidr/add", json=safe_cidrs_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # 4. Add the deny cidr
        deny_cidrs_payload = {"cidrs": ["66.66.1.0/24"]}
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/add", json=deny_cidrs_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # 5. Run worker to consume the job_queue
        await worker.run_once()

        # 6. Get the cidrs in the denylist and ensure they've been split accordingly
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        cidrs_in_denylist = response.json()
        safe_cidr = ipaddress.IPv4Network("66.66.1.0/26")
        deny_cidr = ipaddress.IPv4Network("66.66.1.0/24")
        after_exclusion = sorted([x.compressed for x in deny_cidr.address_exclude(safe_cidr)])
        assert sorted(cidrs_in_denylist) == after_exclusion

        # test that disabled safelists do not affect the result when adding IPs to the denylist
        # 1. Disable the safelist
        list_safe_disabled = list_safe.copy()
        list_safe_disabled["enabled"] = False
        response = await client.put(f"/v1/list/{list_safe['id']}", json=list_safe_disabled, headers=api_token_header)
        assert response.status_code == HTTP_200_OK

        # 2. Add back the deny ip
        deny_cidrs_payload = {"cidrs": ["66.66.1.0/24"]}
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/add", json=deny_cidrs_payload, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED

        # 3. Run worker to consume the job_queue
        await worker.run_once()

        # 4. Get the cidrs in the denylist, now the network didn't split
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert response.json()[0] == "66.66.1.0/24"

        # 1. Test that the denylist cidrs are split again when we re-enable the list
        response = await client.put(f"/v1/list/{list_safe['id']}", json=list_safe, headers=api_token_header)
        assert response.status_code == HTTP_200_OK

        # 2. Run worker to consume the job_queue
        await worker.run_once()

        # 3. Get the cidrs in the denylist and ensure they've been split accordingly
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        cidrs_in_denylist = response.json()
        safe_cidr = ipaddress.IPv4Network("66.66.1.0/26")
        deny_cidr = ipaddress.IPv4Network("66.66.1.0/24")
        after_exclusion = sorted([x.compressed for x in deny_cidr.address_exclude(safe_cidr)])
        assert sorted(cidrs_in_denylist) == after_exclusion

        # Disable the denylist and test that nothing is returned
        list_deny_disabled = list_deny.copy()
        list_deny_disabled["enabled"] = False
        response = await client.put(f"/v1/list/{list_deny['id']}", json=list_deny_disabled, headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 0

        # re-enable and test that cidrs are returned again
        response = await client.put(f"/v1/list/{list_deny['id']}", json=list_deny, headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        # run worker (not required but might catch bugs with update_cleanup
        await worker.run_once()

        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 2
        assert sorted(response.json()) == sorted(["66.66.1.128/25", "66.66.1.64/26"])

        # TEST ADD AND DELETE RAW
        # test delete raw
        delete_raw_cidrs = {
            "cidrs": "hello this is a raw input, I want to delete: 66.66.1.64/26, the following are invalid addresses 32.0 3.2 1.1.1.1/33"
        }
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/delete/raw", json=delete_raw_cidrs, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED
        await worker.run_once()

        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 1
        assert sorted(response.json()) == sorted(["66.66.1.128/25"])

        # test add raw
        add_raw_cidrs = {
            "cidrs": "hello this is a raw input, I want to add again: 66.66.1.64/26, the following are invalid addresses 32.0 3.2 1.1.1.1/33 fasd::dsf:bf"
        }
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/add/raw", json=add_raw_cidrs, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED
        await worker.run_once()

        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 2
        assert sorted(response.json()) == sorted(["66.66.1.128/25", "66.66.1.64/26"])

        # test IP regex
        add_raw_cidrs2 = {
            "cidrs": """
            Invalid addresses:

            12.12.12.1299
            1212.12.12.1299
            hello1.1.1.1bye
            fasd::dsf:bf
            1.1.1.1/33
            2c0f:fb50::/129

            Valid addresses:
            23.23.23.23/32
            13.14.15.16/24 --> raw input is not strict with hosts bits set, this transforns into: 13.14.15.0/24
            2c0f:fb50::/128
            """
        }
        response = await client.post(
            f"/v1/list/{list_deny['id']}/cidr/add/raw", json=add_raw_cidrs2, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED
        await worker.run_once()

        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 5
        assert sorted(response.json()) == sorted(
            ["66.66.1.128/25", "66.66.1.64/26", "23.23.23.23/32", "13.14.15.0/24", "2c0f:fb50::/128"]
        )

        # TEST THAT SPECIAL ADDRESSES NEVER TOUCH THE DB
        list_safe_empty = {"enabled": True, "id": "TEST_CIDRJOB_EMPTY_SAFE1", "list_type": ListTypeEnum.SAFE}
        response = await client.post("/v1/list", json=list_safe_empty, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_safe_empty["id"]
        assert response.json()["list_type"] == list_safe_empty["list_type"]

        invalid_addresses = {"cidrs": ["0.0.0.0", "0.0.0.0/0"]}
        response = await client.post(
            f"/v1/list/{list_safe_empty['id']}/cidr/add", json=invalid_addresses, headers=api_token_header
        )
        assert response.status_code == HTTP_201_CREATED
        await worker.run_once()

        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.SAFE}&list_id={list_safe_empty['id']}",
            headers=api_token_header,
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 0

tests/test_cidr_job.py
aorith/cidr-listings
import asyncio

import pytest
from conftest import get_api_token_header
from litestar.status_codes import HTTP_200_OK, HTTP_201_CREATED, HTTP_400_BAD_REQUEST
from litestar.testing import AsyncTestClient

from app.domain.lists.schemas import ListTypeEnum
from app.lib.scheduled_tasks import TaskDeleteExpired
from app.lib.worker import CidrWorker


@pytest.mark.asyncio
async def test_cidr_ttl(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)
        task_delete_expired = TaskDeleteExpired()
        worker = CidrWorker()

        # Create a Deny list
        list_deny = {"enabled": True, "id": "TEST_CIDR_TTL_DENY1", "list_type": ListTypeEnum.DENY}
        response = await client.post("/v1/list", json=list_deny, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == list_deny["id"]
        assert response.json()["list_type"] == list_deny["list_type"]

        # Try to add an incorrect TTL
        payload = {"cidrs": ["13.1.1.0/24"], "ttl": 0}
        response = await client.post(f"/v1/list/{list_deny['id']}/cidr/add", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_400_BAD_REQUEST
        payload = {"cidrs": ["13.1.1.0/24"], "ttl": -1}
        response = await client.post(f"/v1/list/{list_deny['id']}/cidr/add", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_400_BAD_REQUEST

        # 1. Add a CIDR with 10 TTL
        payload = {"cidrs": ["77.0.1.0/24"], "ttl": 7}
        response = await client.post(f"/v1/list/{list_deny['id']}/cidr/add", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED

        # 2. Run worker and delete task
        await worker.run_once()
        await task_delete_expired.run_once()

        # 3. Ensure the CIDR is still in the list
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) > 0
        assert response.json()[0] == payload["cidrs"][0]

        # 4. Wait longer, until the ttl is expired and then run delete task
        await asyncio.sleep(payload["ttl"] + 1)
        await task_delete_expired.run_once()

        # 5. Ensure the CIDR has been deleted
        response = await client.get(
            f"/v1/cidr/collapsed?list_type={ListTypeEnum.DENY}&list_id={list_deny['id']}", headers=api_token_header
        )
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 0

tests/test_cidr_ttl.py
aorith/cidr-listings
from conftest import get_api_token_header
from litestar.status_codes import (
    HTTP_200_OK,
    HTTP_201_CREATED,
    HTTP_204_NO_CONTENT,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
)
from litestar.testing import AsyncTestClient

from app.domain.lists.schemas import (
    MAX_DESCRIPTION_LEN,
    MAX_LIST_ID_LEN,
    MAX_TAG_LEN,
    ListTypeEnum,
)
from app.lib.settings import get_settings

settings = get_settings()


async def test_list(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        response = await client.get("/v1/list", headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) >= 0

        response = await client.get("/v1/list/NONEXISTENT", headers=api_token_header)
        assert response.status_code == HTTP_404_NOT_FOUND

        response = await client.delete("/v1/list/NONEXISTENT", headers=api_token_header)
        assert response.status_code == HTTP_404_NOT_FOUND


async def test_create_modify(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        test_list = {"id": "MYTESTLIST1", "list_type": "DENY"}

        response = await client.post("/v1/list", json=test_list, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == test_list["id"]
        assert response.json()["list_type"] == test_list["list_type"]

        response = await client.get(f"/v1/list/{test_list['id']}", headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        assert response.json()["id"] == test_list["id"]
        assert response.json()["list_type"] == test_list["list_type"]
        assert "DEFAULT" in response.json()["tags"]

        response = await client.delete(f"/v1/list/{test_list['id']}", headers=api_token_header)
        assert response.status_code == HTTP_204_NO_CONTENT

        response = await client.get(f"/v1/list/{test_list['id']}", headers=api_token_header)
        assert response.status_code == HTTP_404_NOT_FOUND

        test_list2 = {"id": "MYTESTLIST2", "list_type": "SAFE", "tags": ["AA", "BB"]}
        response = await client.post("/v1/list", json=test_list2, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        assert response.json()["id"] == test_list2["id"]
        assert response.json()["list_type"] == test_list2["list_type"]
        for t in test_list2["tags"] + ["DEFAULT"]:
            assert t in response.json()["tags"]

        test_list3 = {"id": "MYTESTLIST3", "list_type": "DENY", "tags": ["AA", "BB"]}
        response = await client.post("/v1/list", json=test_list3, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        response = await client.get(f"/v1/list/{test_list3['id']}", headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        old_created_at = response.json()["created_at"]

        # Assert we cannot modify created_at field or id
        update_list = {"id": "BB", "tags": ["OK"]}
        response = await client.put(f"/v1/list/{test_list3['id']}", json=update_list, headers=api_token_header)
        assert response.status_code == HTTP_200_OK
        assert sorted(response.json()["tags"]) == sorted(update_list["tags"] + ["DEFAULT"])
        assert response.json()["id"] == test_list3["id"]
        assert response.json()["created_at"] == old_created_at


async def test_list_user_scope(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        payload1 = {"login": "testuser01", "password": "abcdefgH1bbcc"}
        response = await client.post("/v1/admin/signup", json=payload1, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED
        payload2 = {"login": "testuser02", "password": "abcdefgH1bbcc"}
        response = await client.post("/v1/admin/signup", json=payload2, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED

        response = await client.post("/v1/auth/token", json=payload1)
        assert response.status_code == HTTP_200_OK
        token1 = f"{response.json()['token_type']} {response.json()['access_token']}"

        response = await client.post("/v1/auth/token", json=payload2)
        assert response.status_code == HTTP_200_OK
        token2 = f"{response.json()['token_type']} {response.json()['access_token']}"

        # Create a list with user1
        response = await client.post(
            "/v1/list", json={"id": "USER1", "list_type": ListTypeEnum.DENY}, headers={"Authorization": token1}
        )
        assert response.status_code == HTTP_201_CREATED
        # Create a list with user2
        response = await client.post(
            "/v1/list", json={"id": "USER2", "list_type": ListTypeEnum.DENY}, headers={"Authorization": token2}
        )
        assert response.status_code == HTTP_201_CREATED

        # Ensure each user only has their list
        response = await client.get("/v1/list", headers={"Authorization": token1})
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 1
        assert response.json()[0]["id"] == "USER1"
        response = await client.get("/v1/list", headers={"Authorization": token2})
        assert response.status_code == HTTP_200_OK
        assert len(response.json()) == 1
        assert response.json()[0]["id"] == "USER2"


async def test_bad_payloads(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        for payload in [
            {"id": "asdf", "list_type": "DENY"},
            {"id": "A" * (MAX_LIST_ID_LEN + 1), "list_type": "DENY"},
            {"id": "AA", "list_type": "DENY", "tags": ["A" * (MAX_TAG_LEN + 1)]},
            {"id": "AAA", "list_type": "DENY", "description": "A" * (MAX_DESCRIPTION_LEN + 1)},
            {"id": "ABC", "list_type": "OOPS"},
            {"id": "1ABC", "list_type": "DENY"},
        ]:
            response = await client.post("/v1/list", json=payload, headers=api_token_header)
            assert response.status_code == HTTP_400_BAD_REQUEST


async def test_cidr_list(test_client: AsyncTestClient) -> None:
    async with test_client as client:
        api_token_header = await get_api_token_header(client)

        payload = {"login": "testcidrlist01", "password": "abcdefgH1bbcc"}
        response = await client.post("/v1/admin/signup", json=payload, headers=api_token_header)
        assert response.status_code == HTTP_201_CREATED

        response = await client.post("/v1/auth/token", json=payload)
        assert response.status_code == HTTP_200_OK
        token = {"Authorization": f"Bearer {response.json()['access_token']}"}

        list_payload = {"id": "TEST_LIST_CIDR", "list_type": "DENY"}
        response = await client.post("/v1/list", json=list_payload, headers=token)
        assert response.status_code == HTTP_201_CREATED

        response = await client.get(f"/v1/list/{list_payload['id']}/cidr", headers=token)
        assert response.status_code == HTTP_200_OK
        assert "cidrs" in response.json()
        assert len(response.json()["cidrs"]) == 0
        assert response.json()["tags"] == ["DEFAULT"]

tests/test_list.py
aorith/cidr-listings