[stalker] init checker

Author: keltecc

checkers/stalker/client.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+
+import json
+import dataclasses
+from typing import Self, Dict, Any
+
+import gornilo.http_clients
+
+import model
+
+
+MAX_BODY_SIZE = 1 << 20 # 1 MiB
+CHUNK_SIZE = 1 << 10 # 1 KiB
+
+TOKEN_HEADER_NAME = 'X-Token'
+
+
+@dataclasses.dataclass
+class Response:
+    code: int
+    content: Dict[str, Any]
+
+
+class Api:
+    def __init__(self: Self, hostname: str, port: int) -> None:
+        self.url = f'http://{hostname}:{port}'
+        self.token = 'x'
+
+    def user_get(self: Self, username: str) -> model.User:
+        url = self.url + f'/users/profile/{username}'
+
+        response = self.http_request('GET', url)
+
+        if isinstance(response, Response) and response.code == 200:
+            return model.User.parse(response.content)
+
+        raise model.ProtocolError('invalid response on user/profile')
+    
+    def user_register(self: Self, username: str, password: str) -> bool:
+        url = self.url + '/users/register'
+        body = {
+            'name': username,
+            'password': password,
+        }
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name == 'AlreadyExistsError':
+            return False
+
+        raise model.ProtocolError('invalid response on user/register')
+
+    def user_login(self: Self, username: str, password: str) -> bool:
+        url = self.url + '/users/login'
+        body = {
+            'name': username,
+            'password': password,
+        }
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name == 'InvalidCredentialsError':
+            return False
+
+        raise model.ProtocolError('invalid response on user/login')
+    
+    def user_logout(self: Self) -> bool:
+        url = self.url + '/users/logout'
+        body = {}
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        raise model.ProtocolError('invalid response on user/logout')
+    
+    def note_get(self: Self, title: str) -> model.Note:
+        url = self.url + f'/notes/{title}'
+
+        response = self.http_request('GET', url)
+
+        if isinstance(response, Response) and response.code == 200:
+            return model.Note.parse(response.content)
+
+        raise model.ProtocolError('invalid response on note/get')
+    
+    def note_create(self: Self, title: str, visible: bool, content: str) -> bool:
+        url = self.url + '/notes'
+        body = {
+            'title': title,
+            'visible': visible,
+            'content': content,
+        }
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name == 'AlreadyExistsError':
+            return False
+
+        raise model.ProtocolError('invalid response on note/create')
+    
+    def note_share(self: Self, title: str, viewer: str) -> bool:
+        url = self.url + f'/notes/{title}/share'
+        body = {
+            'viewer': viewer,
+        }
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name in ('OwnerMismatchError', 'UserNotFoundError'):
+            return False
+
+        raise model.ProtocolError('invalid response on note/share')
+    
+    def note_deny(self: Self, title: str, viewer: str) -> bool:
+        url = self.url + f'/notes/{title}/deny'
+        body = {
+            'viewer': viewer,
+        }
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name in ('OwnerMismatchError', 'UserNotFoundError'):
+            return False
+
+        raise model.ProtocolError('invalid response on note/deny')
+    
+    def note_destroy(self: Self, title: str) -> bool:
+        url = self.url + f'/notes/{title}/destroy'
+        body = {}
+
+        response = self.http_request('POST', url, body)
+
+        if isinstance(response, Response) and response.code == 200:
+            return True
+
+        if isinstance(response, model.ServiceError) and response.name in ('OwnerMismatchError'):
+            return False
+
+        raise model.ProtocolError('invalid response on note/destroy')
+    
+    def http_request(
+            self: Self, method: str, url: str, body: Dict[str, Any] = {},
+    ) -> Response | model.ServiceError:
+        session = gornilo.http_clients.requests_with_retries(
+            status_forcelist = (500, 502),
+        )
+
+        response = session.request(
+            method = method,
+            url = url,
+            json = body,
+            allow_redirects = False,
+            stream = True,
+            headers = {
+                TOKEN_HEADER_NAME: self.token,
+            },
+        )
+
+        self.token = response.headers.get(TOKEN_HEADER_NAME)
+
+        try:
+            content_length = int(response.headers.get('Content-Length', '0'))
+        except Exception:
+            raise model.ProtocolError('invalid http headers')
+
+        if content_length > MAX_BODY_SIZE:
+            raise model.ProtocolError('body size is too big')
+
+        chunks = []
+        total_length = 0
+
+        for chunk in response.iter_content(CHUNK_SIZE, decode_unicode = False):
+            chunks.append(chunk)
+            total_length += len(chunk)
+
+            if total_length > MAX_BODY_SIZE:
+                raise model.ProtocolError('body size is too big')
+
+        data = b''.join(chunks)
+
+        try:
+            content = json.loads(data)
+        except Exception:
+            raise model.ProtocolError('failed to parse json response')
+        
+        if 'error' in content:
+            error = content['error']
+
+            if not isinstance(error, dict):
+                raise model.ProtocolError('failed to parse error')
+
+            return model.ServiceError.parse(error)
+
+        return Response(response.status_code, content)

checkers/stalker/model.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+
+import dataclasses
+from typing import List, Dict, Any
+
+
+class ProtocolError(Exception):
+    pass
+
+
+class ValidationError(Exception):
+    pass
+
+
+@dataclasses.dataclass
+class ServiceError:
+    name: str
+    message: str | None
+
+    @staticmethod
+    def parse(obj: Dict[str, Any]) -> 'ServiceError':
+        name = obj.get('name')
+        message = obj.get('message')
+
+        assertions = [
+            isinstance(name, str),
+            any([
+                message is None,
+                isinstance(message, str),
+            ]),
+        ]
+
+        if not all(assertions):
+            raise ValidationError('invalid error structure')
+
+        return ServiceError(name = name, message = message)
+
+
+@dataclasses.dataclass
+class Note:
+    title: str
+    visible: bool
+    content: str | None
+    viewers: List[str] | None
+    owner: str
+
+    @staticmethod
+    def parse(obj: Dict[str, Any]) -> 'Note':
+        title = obj.get('title')
+        visible = obj.get('visible')
+        content = obj.get('content')
+        viewers = obj.get('viewers')
+        owner = obj.get('owner')
+
+        assertions = [
+            isinstance(title, str),
+            isinstance(visible, bool),
+            any([
+                content is None,
+                isinstance(content, str),
+            ]),
+            any([
+                viewers is None,
+                all([
+                    isinstance(viewers, list),
+                    all(isinstance(viewer, str) for viewer in viewers),
+                ]),
+            ]),
+            isinstance(owner, str),
+        ]
+
+        if not all(assertions):
+            raise ValidationError('invalid note structure')
+        
+        return Note(
+            title = title,
+            visible = visible,
+            content = content,
+            viewers = viewers,
+            owner = owner,
+        )
+
+
+@dataclasses.dataclass
+class User:
+    name: str
+    owned_notes: List[str]
+    shared_notes: List[str] | None
+
+    @staticmethod
+    def parse(obj: Dict[str, Any]) -> 'User':
+        name = obj.get('name')
+        owned_notes = obj.get('ownedNotes')
+        shared_notes = obj.get('sharedNotes')
+
+        assertions = [
+            isinstance(name, str),
+            all([
+                isinstance(owned_notes, list),
+                all(isinstance(owned_note, str) for owned_note in owned_notes),
+            ]),
+            any([
+                shared_notes is None,
+                all([
+                    isinstance(shared_notes, list),
+                    all(isinstance(shared_note, str) for shared_note in shared_notes),
+                ]),
+            ]),
+        ]
+
+        if not all(assertions):
+            raise ValidationError('invalid user structure')
+
+        return User(
+            name = name,
+            owned_notes = owned_notes,
+            shared_notes = shared_notes,
+        )

checkers/stalker/requirements.txt

@@ -0,0 +1,2 @@
+gornilo==0.9.3
+requests==2.28.2