Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
f1911d9b15 | ||
eedc1d5679 | |||
1825e3109c | |||
bfd30324e1 | |||
750f79f03a |
@ -1,2 +1,3 @@
|
|||||||
from bump.bump import Bump
|
from bump.bump import Bump
|
||||||
|
import bump.cli as cli
|
||||||
|
|
||||||
|
125
bump/bump.py
Normal file → Executable file
125
bump/bump.py
Normal file → Executable file
@ -6,6 +6,7 @@ import requests
|
|||||||
import base64
|
import base64
|
||||||
import re
|
import re
|
||||||
import os
|
import os
|
||||||
|
import typer
|
||||||
from random_word import RandomWords
|
from random_word import RandomWords
|
||||||
|
|
||||||
SENDER_LENGTH = 4
|
SENDER_LENGTH = 4
|
||||||
@ -51,22 +52,57 @@ def password_decrypt(token: bytes, password: str) -> bytes:
|
|||||||
return Fernet(key).decrypt(token)
|
return Fernet(key).decrypt(token)
|
||||||
|
|
||||||
class Bump:
|
class Bump:
|
||||||
def __init__(self, secret=None, secrets_file=None):
|
def __init__(self, secret: str = None) -> None:
|
||||||
if None == secrets_file:
|
self._initialize_parameters()
|
||||||
secrets_file = os.path.join(os.path.expanduser('~'), '.config/bump/secrets_file')
|
self._save_or_load_secrets(secret)
|
||||||
self.secrets = self._load_secrets(secret, secrets_file)
|
|
||||||
|
|
||||||
|
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 == []:
|
if self.secrets == []:
|
||||||
print("you seem to not have a secret in your secrets file! Creating one now...")
|
print("you seem to not have a secret in your secrets file! Creating one now...")
|
||||||
self.generate_secret(secrets_file)
|
self.generate_secret()
|
||||||
self.show_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.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:
|
try:
|
||||||
with open(secrets_file, 'r') as f:
|
with open(self.secrets_file, 'r') as f:
|
||||||
secrets = f.read().splitlines()
|
secrets = f.read().splitlines()
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
secrets = []
|
secrets = []
|
||||||
@ -75,30 +111,24 @@ class Bump:
|
|||||||
|
|
||||||
return secrets
|
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]+$')
|
pattern = re.compile('^[a-zA-Z]+$')
|
||||||
WORDS = r.get_random_words()
|
|
||||||
secret = ""
|
secret = ""
|
||||||
word_count = 0
|
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)
|
word = r.get_random_word()
|
||||||
print(pattern.match(word))
|
if pattern.match(word) and len(word) < 10 and len(word) >= 4:
|
||||||
if pattern.match(word) and len(word) < 10:
|
|
||||||
secret += word + "-"
|
secret += word + "-"
|
||||||
word_count += 1
|
word_count += 1
|
||||||
|
|
||||||
secret = secret[:-1]
|
secret = secret[:-1]
|
||||||
|
return secret
|
||||||
|
|
||||||
self.secrets.append(secret)
|
def show_secret(self) -> None:
|
||||||
|
|
||||||
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):
|
|
||||||
print("Scan this QR Code with the Bump app to connect")
|
print("Scan this QR Code with the Bump app to connect")
|
||||||
for secret in self.secrets:
|
for secret in self.secrets:
|
||||||
qr = qrcode.QRCode(
|
qr = qrcode.QRCode(
|
||||||
@ -112,24 +142,24 @@ class Bump:
|
|||||||
print("")
|
print("")
|
||||||
print(secret)
|
print(secret)
|
||||||
|
|
||||||
def _get_password(self, index=0):
|
def _get_password(self, index: int = 0) -> str:
|
||||||
secret = self.secrets[index].split('-')
|
secret = self.secrets[index].split('-')
|
||||||
return "-".join(secret[SENDER_LENGTH:])
|
return "-".join(secret[SENDER_LENGTH:])
|
||||||
|
|
||||||
|
|
||||||
def _get_sender(self, index=0):
|
def _get_sender(self, index: int = 0) -> str:
|
||||||
words = self.secrets[index].split('-')
|
words = self.secrets[index].split('-')
|
||||||
return "-".join(words[0:SENDER_LENGTH])
|
return "-".join(words[0:SENDER_LENGTH])
|
||||||
|
|
||||||
def _encrypt(self, data):
|
def _encrypt(self, data:str) -> str:
|
||||||
password = self._get_password()
|
password = self._get_password()
|
||||||
return password_encrypt(data.encode(), password)
|
return password_encrypt(data.encode(), password)
|
||||||
|
|
||||||
def _decrypt(self, data):
|
def _decrypt(self, data: str) -> str:
|
||||||
password = self._get_password()
|
password = self._get_password()
|
||||||
return password_decrypt(data, password).decode()
|
return password_decrypt(data, password).decode()
|
||||||
|
|
||||||
def push(self, title='', data=''):
|
def push(self, title='', data='') -> None:
|
||||||
params = {
|
params = {
|
||||||
'sender': self._get_sender(),
|
'sender': self._get_sender(),
|
||||||
'title': title,
|
'title': title,
|
||||||
@ -137,22 +167,31 @@ class Bump:
|
|||||||
}
|
}
|
||||||
return self._set_post("push", params)
|
return self._set_post("push", params)
|
||||||
|
|
||||||
def peek(self):
|
def peek(self) -> dict:
|
||||||
return self._get_post("peek")
|
return self._get_post("peek")
|
||||||
|
|
||||||
def pop(self):
|
def pop(self) -> dict:
|
||||||
return self._get_post("pop")
|
return self._get_post("pop")
|
||||||
|
|
||||||
def list(self):
|
def list(self) -> list:
|
||||||
params = {
|
params = {
|
||||||
'minutes': 2
|
'minutes': 2
|
||||||
}
|
}
|
||||||
return self._get_post("list", params)
|
return self._get_post("list", params)
|
||||||
|
|
||||||
|
|
||||||
def clear(self):
|
def clear(self) -> None:
|
||||||
return self._get_post("clear")
|
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):
|
def delete_sender(self):
|
||||||
return self._set_post("delete_sender")
|
return self._set_post("delete_sender")
|
||||||
|
|
||||||
@ -168,7 +207,7 @@ class Bump:
|
|||||||
messages = requests.post(url, params).json()
|
messages = requests.post(url, params).json()
|
||||||
print(messages)
|
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
|
url = self.URL + mechanism
|
||||||
|
|
||||||
params = {
|
params = {
|
||||||
@ -193,29 +232,25 @@ class Bump:
|
|||||||
return [messages]
|
return [messages]
|
||||||
|
|
||||||
|
|
||||||
def _load_log(self):
|
def _load_log(self) -> list:
|
||||||
with open('.bump_log', "r+") as f:
|
with open('.bump_log', "r+") as f:
|
||||||
return f.readlines()
|
return f.readlines()
|
||||||
|
|
||||||
def _save_log(self, message):
|
def _save_log(self, message: str) -> None:
|
||||||
with open('.bump_log', "a") as f:
|
with open('.bump_log', "a") as f:
|
||||||
f.write(self._to_log_line(message))
|
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"
|
return json.dumps(message) + "\n"
|
||||||
|
|
||||||
def alert(self, sleep_time=1):
|
def alert(self, sleep_time: int = 1) -> None:
|
||||||
log = self._load_log()
|
log = self._load_log()
|
||||||
while True:
|
while True:
|
||||||
time.sleep(sleep_time)
|
time.sleep(sleep_time)
|
||||||
messages = self.list()
|
messages = self.list()
|
||||||
for message in messages:
|
for message in messages['messages']:
|
||||||
if self._to_log_line(message) not in log and message != {}:
|
if self._to_log_line(message) not in log and message != {}:
|
||||||
print(message['data'])
|
print({'title': message['title'], 'data': message['data']})
|
||||||
self._save_log(message)
|
self._save_log(message)
|
||||||
log = self.load_log()
|
log = self._load_log()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
b = Bump()
|
|
||||||
|
60
bump/cli.py
Normal file
60
bump/cli.py
Normal file
@ -0,0 +1,60 @@
|
|||||||
|
from typing import Optional
|
||||||
|
import bump
|
||||||
|
import typer
|
||||||
|
import sys
|
||||||
|
import select
|
||||||
|
|
||||||
|
|
||||||
|
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:Optional[bool] = typer.Option(False, help="remove all senders"),
|
||||||
|
alert:Optional[bool] = typer.Option(False, help="check continuously for new messages"),
|
||||||
|
secret:Optional[str] = typer.Option(None, help="change the sender and e2e keys")):
|
||||||
|
|
||||||
|
bp = bump.Bump()
|
||||||
|
|
||||||
|
pipe = None
|
||||||
|
if select.select([sys.stdin,],[],[],0.0)[0]:
|
||||||
|
pipe = sys.stdin.readline()
|
||||||
|
if message_arg is not None:
|
||||||
|
push(message_arg)
|
||||||
|
elif pipe is not None:
|
||||||
|
push(pipe)
|
||||||
|
elif message is not None:
|
||||||
|
push(message)
|
||||||
|
elif secret is not None:
|
||||||
|
secrets(secret)
|
||||||
|
elif reset:
|
||||||
|
delete_senders()
|
||||||
|
elif alert:
|
||||||
|
alert_process()
|
||||||
|
else:
|
||||||
|
info()
|
||||||
|
|
||||||
|
def delete_senders():
|
||||||
|
bp = bump.Bump()
|
||||||
|
bp.delete_senders()
|
||||||
|
|
||||||
|
def alert_process():
|
||||||
|
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)
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
10
setup.py
10
setup.py
@ -1,10 +1,15 @@
|
|||||||
from setuptools import setup
|
from setuptools import setup
|
||||||
|
|
||||||
setup(
|
setup(
|
||||||
|
entry_points={
|
||||||
|
'console_scripts': [
|
||||||
|
'bump = bump.cli:main',
|
||||||
|
],
|
||||||
|
},
|
||||||
name='bump_python',
|
name='bump_python',
|
||||||
packages=['bump'],
|
packages=['bump'],
|
||||||
description='Notify your Phone from Python',
|
description='Notify your Phone from Python',
|
||||||
version='0.1.1',
|
version='0.1.5',
|
||||||
url='https://git.maenle.tech/raphael/bump_python',
|
url='https://git.maenle.tech/raphael/bump_python',
|
||||||
author='Raphael Maenle',
|
author='Raphael Maenle',
|
||||||
author_email='raphael@maenle.net',
|
author_email='raphael@maenle.net',
|
||||||
@ -18,6 +23,7 @@ setup(
|
|||||||
'qrcode',
|
'qrcode',
|
||||||
'pyyaml',
|
'pyyaml',
|
||||||
'random-word',
|
'random-word',
|
||||||
'cryptography'
|
'cryptography',
|
||||||
|
'typer'
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
Loading…
Reference in New Issue
Block a user