senders can now be added and removed via cli

- --reset flag (default false) can now be set
  to remove all senders
- multiple senders can be added now
This commit is contained in:
Raphael Maenle 2022-01-17 11:44:43 +01:00
parent 750f79f03a
commit bfd30324e1
2 changed files with 112 additions and 44 deletions

View File

@ -52,21 +52,57 @@ def password_decrypt(token: bytes, password: str) -> bytes:
return Fernet(key).decrypt(token)
class Bump:
def __init__(self, secret=None, secrets_file=None):
if None == secrets_file:
secrets_file = os.path.join(os.path.expanduser('~'), '.config/bump/secrets_file')
self.secrets = self._load_secrets(secret, secrets_file)
def __init__(self, secret: str = None) -> None:
self._initialize_parameters()
self._save_or_load_secrets(secret)
def _save_or_load_secrets(self, secret: str = None) -> None:
self._check_and_save_secret(secret)
self.secrets = self._load_secrets(secret)
if self.secrets == []:
print("you seem to not have a secret in your secrets file! Creating one now...")
self.generate_secret(secrets_file)
self.generate_secret()
def _check_secret_valid(self, secret: str) -> bool:
pattern = re.compile('^[a-zA-Z-]+$')
words = secret.split('-')
if (pattern.match(secret) and
len(words) >= SENDER_LENGTH + PASSWORD_LENGTH and
sum([len(word) for word in words[SENDER_LENGTH:]]) > 32):
return True
else:
return False
def _initialize_parameters(self) -> None:
self.secrets_file = os.path.join(os.path.expanduser('~'), '.config/bump/secrets_file')
self.URL = "https://bump.maenle.net/api/"
self.secrets = []
def _check_and_save_secret(self, secret:str = None) -> None:
if secret is not None:
if self._check_secret_valid(secret):
self._save_secret(secret)
else:
print("--------------")
print("invalid secret")
print("--------------")
def _load_secrets(self, secret, secrets_file):
def _save_secret(self, secret: str) -> None:
self.secrets.append(secret)
if not os.path.exists(os.path.dirname(self.secrets_file)):
os.makedirs(os.path.dirname(self.secrets_file))
with open(self.secrets_file, 'a+') as f:
f.write(secret + '\n')
def _load_secrets(self, secret: str) -> list:
try:
with open(secrets_file, 'r') as f:
with open(self.secrets_file, 'r') as f:
secrets = f.read().splitlines()
except FileNotFoundError:
secrets = []
@ -75,30 +111,25 @@ class Bump:
return secrets
def generate_secret(self, secrets_file):
def generate_secret(self):
keywords = self._generate_keywords()
self._check_and_save_secret(keywords)
def _generate_keywords(self) -> str:
pattern = re.compile('^[a-zA-Z]+$')
WORDS = r.get_random_words()
secret = ""
word_count = 0
while word_count < SENDER_LENGTH + PASSWORD_LENGTH or len(secret[SENDER_LENGTH:]) < 32:
while word_count < SENDER_LENGTH + PASSWORD_LENGTH:
word = random.choice(WORDS)
print(pattern.match(word))
if pattern.match(word) and len(word) < 10:
if pattern.match(word) and len(word) < 10 and len(word) >= 4:
secret += word + "-"
word_count += 1
secret = secret[:-1]
return secret
self.secrets.append(secret)
if not os.path.exists(os.path.dirname(secrets_file)):
os.makedirs(os.path.dirname(secrets_file))
with open(secrets_file, 'a+') as f:
f.write(secret + '\n')
def show_secret(self):
def show_secret(self) -> None:
print("Scan this QR Code with the Bump app to connect")
for secret in self.secrets:
qr = qrcode.QRCode(
@ -112,24 +143,24 @@ class Bump:
print("")
print(secret)
def _get_password(self, index=0):
def _get_password(self, index: int = 0) -> str:
secret = self.secrets[index].split('-')
return "-".join(secret[SENDER_LENGTH:])
def _get_sender(self, index=0):
def _get_sender(self, index: int = 0) -> str:
words = self.secrets[index].split('-')
return "-".join(words[0:SENDER_LENGTH])
def _encrypt(self, data):
def _encrypt(self, data:str) -> str:
password = self._get_password()
return password_encrypt(data.encode(), password)
def _decrypt(self, data):
def _decrypt(self, data: str) -> str:
password = self._get_password()
return password_decrypt(data, password).decode()
def push(self, title='', data=''):
def push(self, title='', data='') -> None:
params = {
'sender': self._get_sender(),
'title': title,
@ -137,22 +168,31 @@ class Bump:
}
return self._set_post("push", params)
def peek(self):
def peek(self) -> dict:
return self._get_post("peek")
def pop(self):
def pop(self) -> dict:
return self._get_post("pop")
def list(self):
def list(self) -> list:
params = {
'minutes': 2
}
return self._get_post("list", params)
def clear(self):
def clear(self) -> None:
return self._get_post("clear")
def delete_senders(self):
self.secrets = []
if not os.path.exists(os.path.dirname(self.secrets_file)):
os.makedirs(os.path.dirname(self.secrets_file))
with open(self.secrets_file, 'w+') as f:
f.write("")
def delete_sender(self):
return self._set_post("delete_sender")
@ -168,7 +208,7 @@ class Bump:
messages = requests.post(url, params).json()
print(messages)
def _get_post(self, mechanism, add_params = None):
def _get_post(self, mechanism: str, add_params: dict = None) -> dict:
url = self.URL + mechanism
params = {
@ -193,18 +233,18 @@ class Bump:
return [messages]
def _load_log(self):
def _load_log(self) -> list:
with open('.bump_log', "r+") as f:
return f.readlines()
def _save_log(self, message):
def _save_log(self, message: str) -> None:
with open('.bump_log', "a") as f:
f.write(self._to_log_line(message))
def _to_log_line(self, message):
def _to_log_line(self, message: str) -> str:
return json.dumps(message) + "\n"
def alert(self, sleep_time=1):
def alert(self, sleep_time: int = 1) -> None:
log = self._load_log()
while True:
time.sleep(sleep_time)

View File

@ -1,20 +1,48 @@
from typing import Optional
import bump
import typer
cli = typer.Typer()
@cli.command()
def default(message = typer.Argument(None, help="The message to be bumped")):
def default(
message_arg:Optional[str] = typer.Argument(None, help="The message to be bumped"),
message:Optional[str] = typer.Option(None, help="The message to be bumped"),
reset:bool = typer.Option(False, help="remove all senders"),
alert:Optional[str] = typer.Option(None, help="check continuously for new messages"),
secret:Optional[str] = typer.Option(None, help="change the sender and e2e keys")):
bp = bump.Bump()
if message is None:
bp.show_secret()
if message_arg is not None:
push(message_arg)
elif message is not None:
push(message)
elif secret is not None:
secrets(secret)
elif reset:
delete_senders()
elif alert is not None:
alert()
else:
bp.push(message)
info()
@cli.command()
def push():
def delete_senders():
bp = bump.Bump()
bp.version()
bp.delete_senders()
def alert():
bp = bump.Bump()
bp.alert()
def secrets(secret:str):
if secret is not None:
bp = bump.Bump(secret)
info()
def push(message:str):
bp = bump.Bump()
bp.push(message)
def info():
bp = bump.Bump()
bp.show_secret()
def main():
typer.run(default)