client/securedrop_client/sdk/sdlocalobjects.py (149 lines of code) (raw):

class BaseError(Exception): """For generic errors not covered by other exceptions""" def __init__(self, message: str | None = None) -> None: self.msg = message def __str__(self) -> str: return repr(self.msg) class ReplyError(BaseError): "For errors on reply messages" def __init__(self, message: str) -> None: self.msg = message def __str__(self) -> str: return repr(self.msg) class WrongUUIDError(BaseError): "For missing UUID, can be for source or submission" def __init__(self, message: str) -> None: self.msg = message def __str__(self) -> str: return repr(self.msg) class AuthError(BaseError): "For Authentication errors" def __init__(self, message: str) -> None: self.msg = message def __str__(self) -> str: return repr(self.msg) class AttributeError(BaseError): def __init__(self, message: str) -> None: self.msg = message def __str__(self) -> str: return repr(self.msg) class Reply: """ This class represents a reply to the source. """ def __init__(self, **kwargs) -> None: # type: ignore self.filename = "" # type: str self.journalist_uuid = "" # type: str self.journalist_username = "" # type: str self.journalist_first_name = "" # type: str self.journalist_last_name = "" # type: str self.is_deleted_by_source = False # type: bool self.reply_url = "" # type: str self.size = 0 # type: int self.source_url = "" # type: str self.source_uuid = "" # type: str self.uuid = "" # type: str if {"uuid", "filename"} == set(kwargs.keys()): # Then we are creating an object for fetching from the server. self.uuid = kwargs["uuid"] self.filename = kwargs["filename"] return # Fetch an object only by uuid and soure_uuid. elif {"uuid", "source_uuid"} == set(kwargs.keys()): self.uuid = kwargs["uuid"] self.source_uuid = kwargs["source_uuid"] return try: self.filename = kwargs["filename"] self.journalist_uuid = kwargs["journalist_uuid"] self.journalist_username = kwargs["journalist_username"] self.journalist_first_name = kwargs["journalist_first_name"] self.journalist_last_name = kwargs["journalist_last_name"] self.is_deleted_by_source = kwargs["is_deleted_by_source"] self.reply_url = kwargs["reply_url"] self.size = kwargs["size"] self.source_url = kwargs["source_url"] self.uuid = kwargs["uuid"] self.seen_by = kwargs["seen_by"] except KeyError as err: raise AttributeError(f"Missing key {err.args[0]}") from err # Now let us set source uuid values = self.source_url.split("/") self.source_uuid = values[-1] class Submission: """ This class represents a submission object in the server. """ def __init__(self, **kwargs) -> None: # type: ignore self.download_url = "" # type: str self.filename = "" # type: str self.is_read = False # type: bool self.size = 0 # type: int self.source_url = "" # type: str self.source_uuid = "" # type: str self.submission_url = "" # type: str self.uuid = "" # type: str if list(kwargs.keys()) == ["uuid"]: # Means we are creating an object only for fetching from server. self.uuid = kwargs["uuid"] return elif list(kwargs.keys()) == ["uuid", "source_uuid"]: self.uuid = kwargs["uuid"] self.source_uuid = kwargs["source_uuid"] return try: self.download_url = kwargs["download_url"] self.filename = kwargs["filename"] self.is_read = kwargs["is_read"] self.size = kwargs["size"] self.source_url = kwargs["source_url"] self.submission_url = kwargs["submission_url"] self.uuid = kwargs["uuid"] self.seen_by = kwargs["seen_by"] except KeyError as err: raise AttributeError(f"Missing key {err.args[0]}") from err _, self.source_uuid = self.source_url.rsplit("/", 1) def is_file(self) -> bool: return self.filename.endswith("doc.gz.gpg") or self.filename.endswith("doc.zip.gpg") class Source: """ This class represents a source object in the server. """ def __init__(self, **kwargs) -> None: # type: ignore self.add_star_url = "" # type: str self.interaction_count = 0 # type: int self.is_flagged = False # type: bool self.is_starred = False # type: bool self.journalist_designation = "" # type: str self.key: dict = {} self.last_updated = "" # type: str self.number_of_documents = 0 # type: int self.number_of_messages = 0 # type: int self.remove_star_url = "" # type: str self.replies_url = "" # type: str self.submissions_url = "" # type: str self.url = "" # type: str self.uuid = "" # type: str if list(kwargs.keys()) == ["uuid"]: # Means we are creating an object only for fetching from server. self.uuid = kwargs["uuid"] return for key in [ "add_star_url", "interaction_count", "is_flagged", "is_starred", "journalist_designation", "key", "last_updated", "number_of_documents", "number_of_messages", "remove_star_url", "replies_url", "submissions_url", "url", "uuid", ]: if key not in kwargs: AttributeError(f"Missing key {key}") setattr(self, key, kwargs[key]) class User: """ This class represents a user (journalist or admin) of the Journalist Interface. """ def __init__(self, **kwargs) -> None: # type: ignore self.first_name = "" # type: str self.last_name = "" # type: str self.username = "" # type: str self.uuid = "" # type: str for key in [ "first_name", "last_name", "username", "uuid", ]: if key not in kwargs: AttributeError(f"Missing key {key}") setattr(self, key, kwargs[key])