From a96f6bd06733199cf4af3b05acdce1f1d43c85c2 Mon Sep 17 00:00:00 2001 From: inexcode Date: Thu, 11 Aug 2022 23:11:00 +0400 Subject: [PATCH] LInting --- .pylintrc | 1 + selfprivacy_api/actions/api_tokens.py | 9 +-- selfprivacy_api/graphql/queries/system.py | 5 +- .../services/bitwarden/__init__.py | 2 +- .../services/generic_service_mover.py | 4 +- .../services/mailserver/__init__.py | 10 ++-- selfprivacy_api/services/service.py | 59 +++++++++++++------ selfprivacy_api/utils/network.py | 8 +-- 8 files changed, 55 insertions(+), 43 deletions(-) diff --git a/.pylintrc b/.pylintrc index c6d73d8..9135ea9 100644 --- a/.pylintrc +++ b/.pylintrc @@ -1,2 +1,3 @@ [MASTER] init-hook="from pylint.config import find_pylintrc; import os, sys; sys.path.append(os.path.dirname(find_pylintrc()))" +extension-pkg-whitelist=pydantic diff --git a/selfprivacy_api/actions/api_tokens.py b/selfprivacy_api/actions/api_tokens.py index 3a57ef7..61c695d 100644 --- a/selfprivacy_api/actions/api_tokens.py +++ b/selfprivacy_api/actions/api_tokens.py @@ -3,6 +3,7 @@ from datetime import datetime from typing import Optional from pydantic import BaseModel + from selfprivacy_api.utils.auth import ( delete_token, generate_recovery_token, @@ -42,14 +43,10 @@ def get_api_tokens_with_caller_flag(caller_token: str) -> list[TokenInfoWithIsCa class NotFoundException(Exception): """Not found exception""" - pass - class CannotDeleteCallerException(Exception): """Cannot delete caller exception""" - pass - def delete_api_token(caller_token: str, token_name: str) -> None: """Delete the token""" @@ -98,14 +95,10 @@ def get_api_recovery_token_status() -> RecoveryTokenStatus: class InvalidExpirationDate(Exception): """Invalid expiration date exception""" - pass - class InvalidUsesLeft(Exception): """Invalid uses left exception""" - pass - def get_new_api_recovery_key( expiration_date: Optional[datetime] = None, uses_left: Optional[int] = None diff --git a/selfprivacy_api/graphql/queries/system.py b/selfprivacy_api/graphql/queries/system.py index b5fb6a6..decb3f0 100644 --- a/selfprivacy_api/graphql/queries/system.py +++ b/selfprivacy_api/graphql/queries/system.py @@ -75,14 +75,13 @@ def get_ssh_settings() -> SshSettings: return SshSettings( enable=settings.enable, password_authentication=settings.passwordAuthentication, - root_ssh_keys=settings.rootSshKeys, + root_ssh_keys=settings.rootKeys, ) def get_system_timezone() -> str: """Get system timezone""" - with ReadUserData() as user_data: - return system_actions.get_timezone() + return system_actions.get_timezone() @strawberry.type diff --git a/selfprivacy_api/services/bitwarden/__init__.py b/selfprivacy_api/services/bitwarden/__init__.py index 21ba40a..599e0e4 100644 --- a/selfprivacy_api/services/bitwarden/__init__.py +++ b/selfprivacy_api/services/bitwarden/__init__.py @@ -35,7 +35,7 @@ class Bitwarden(Service): return "Bitwarden is a password manager." @staticmethod - def get_svg_icon(self) -> str: + def get_svg_icon() -> str: """Read SVG icon from file and return it as base64 encoded string.""" with open("selfprivacy_api/services/bitwarden/bitwarden.svg", "rb") as f: return base64.b64encode(f.read()).decode("utf-8") diff --git a/selfprivacy_api/services/generic_service_mover.py b/selfprivacy_api/services/generic_service_mover.py index f625f99..1a9c0ac 100644 --- a/selfprivacy_api/services/generic_service_mover.py +++ b/selfprivacy_api/services/generic_service_mover.py @@ -1,9 +1,7 @@ """Generic handler for moving services""" -import base64 import subprocess import time -import typing import pathlib import shutil @@ -149,7 +147,7 @@ def move_service( progress=20, ) current_progress = 20 - folder_percentage = 50 / len(folder_names) + folder_percentage = 50 // len(folder_names) for folder in folder_names: shutil.move( f"/volumes/{old_volume}/{folder.name}", diff --git a/selfprivacy_api/services/mailserver/__init__.py b/selfprivacy_api/services/mailserver/__init__.py index c97864d..3eaf8c6 100644 --- a/selfprivacy_api/services/mailserver/__init__.py +++ b/selfprivacy_api/services/mailserver/__init__.py @@ -120,8 +120,8 @@ class MailServer(Service): @staticmethod def get_dns_records() -> typing.List[ServiceDnsRecord]: - dkim_record = get_dkim_key() domain = get_domain() + dkim_record = get_dkim_key(domain) ip4 = get_ip4() if dkim_record is None: @@ -129,16 +129,16 @@ class MailServer(Service): return [ ServiceDnsRecord( - type="MX", name=domain, data=domain, ttl=3600, priority=10 + type="MX", name=domain, content=domain, ttl=3600, priority=10 ), ServiceDnsRecord( - type="TXT", name="_dmarc", data=f"v=DMARC1; p=none", ttl=3600 + type="TXT", name="_dmarc", content=f"v=DMARC1; p=none", ttl=3600 ), ServiceDnsRecord( - type="TXT", name=domain, data=f"v=spf1 a mx ip4:{ip4} -all", ttl=3600 + type="TXT", name=domain, content=f"v=spf1 a mx ip4:{ip4} -all", ttl=3600 ), ServiceDnsRecord( - type="TXT", name="selector._domainkey", data=dkim_record, ttl=3600 + type="TXT", name="selector._domainkey", content=dkim_record, ttl=3600 ), ] diff --git a/selfprivacy_api/services/service.py b/selfprivacy_api/services/service.py index 43c4fbb..5627e17 100644 --- a/selfprivacy_api/services/service.py +++ b/selfprivacy_api/services/service.py @@ -3,6 +3,8 @@ from abc import ABC, abstractmethod from enum import Enum import typing +from pydantic import BaseModel + from selfprivacy_api.utils.block_devices import BlockDevice @@ -16,12 +18,12 @@ class ServiceStatus(Enum): OFF = "OFF" -class ServiceDnsRecord: +class ServiceDnsRecord(BaseModel): type: str name: str content: str ttl: int - priority: typing.Optional[int] + priority: typing.Optional[int] = None class Service(ABC): @@ -30,80 +32,99 @@ class Service(ABC): can be installed, configured and used by a user. """ + @staticmethod @abstractmethod - def get_id(self) -> str: + def get_id() -> str: pass + @staticmethod @abstractmethod - def get_display_name(self) -> str: + def get_display_name() -> str: pass + @staticmethod @abstractmethod - def get_description(self) -> str: + def get_description() -> str: pass + @staticmethod @abstractmethod - def get_svg_icon(self) -> str: + def get_svg_icon() -> str: pass + @staticmethod @abstractmethod def is_movable() -> bool: pass + @staticmethod @abstractmethod def is_required() -> bool: pass + @staticmethod @abstractmethod - def is_enabled(self) -> bool: + def is_enabled() -> bool: pass + @staticmethod @abstractmethod - def get_status(self) -> ServiceStatus: + def get_status() -> ServiceStatus: pass + @staticmethod @abstractmethod - def enable(self): + def enable(): pass + @staticmethod @abstractmethod - def disable(self): + def disable(): pass + @staticmethod @abstractmethod - def stop(self): + def stop(): pass + @staticmethod @abstractmethod - def start(self): + def start(): pass + @staticmethod @abstractmethod - def restart(self): + def restart(): pass + @staticmethod @abstractmethod - def get_configuration(self): + def get_configuration(): pass + @staticmethod @abstractmethod - def set_configuration(self, config_items): + def set_configuration(config_items): pass + @staticmethod @abstractmethod - def get_logs(self): + def get_logs(): pass + @staticmethod @abstractmethod - def get_storage_usage(self) -> int: + def get_storage_usage() -> int: pass + @staticmethod @abstractmethod - def get_dns_records(self) -> typing.List[ServiceDnsRecord]: + def get_dns_records() -> typing.List[ServiceDnsRecord]: pass + @staticmethod @abstractmethod - def get_location(self) -> str: + def get_location() -> str: pass @abstractmethod diff --git a/selfprivacy_api/utils/network.py b/selfprivacy_api/utils/network.py index 9b1ba99..c1b8a2b 100644 --- a/selfprivacy_api/utils/network.py +++ b/selfprivacy_api/utils/network.py @@ -5,7 +5,7 @@ import re from typing import Optional -def get_ip4() -> Optional[str]: +def get_ip4() -> str: """Get IPv4 address""" try: ip4 = subprocess.check_output(["ip", "addr", "show", "dev", "eth0"]).decode( @@ -14,10 +14,10 @@ def get_ip4() -> Optional[str]: ip4 = re.search(r"inet (\d+\.\d+\.\d+\.\d+)\/\d+", ip4) except subprocess.CalledProcessError: ip4 = None - return ip4.group(1) if ip4 else None + return ip4.group(1) if ip4 else "" -def get_ip6() -> Optional[str]: +def get_ip6() -> str: """Get IPv6 address""" try: ip6 = subprocess.check_output(["ip", "addr", "show", "dev", "eth0"]).decode( @@ -26,4 +26,4 @@ def get_ip6() -> Optional[str]: ip6 = re.search(r"inet6 (\S+)\/\d+", ip6) except subprocess.CalledProcessError: ip6 = None - return ip6.group(1) if ip6 else None + return ip6.group(1) if ip6 else ""