nogafam
/
ig-pixelfed-mirror
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Added wildcard login/logout + argv bind addr + improved code

This commit is contained in:
Bofh 2021-02-10 19:07:03 +01:00
parent babaa29e74
commit 29dea31a0b
3 changed files with 100 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -127,6 +127,10 @@ Optionally, you can set a **binding port** with a positional argument like this:
`curl 127.0.0.1:8080/list`
**List** accounts in a pretty HTML/CSS interface:
`curl 127.0.0.1:8080/mirrors`
**Add a new Instagram account to mirror**. This is done **syncronously until it finishes adding** account information, once it's done, it **calls update asyncronously**.
`curl 127.0.0.1:8080/<username>/add`

48
igmirror.py
View File

@ -134,21 +134,49 @@ def delete_statuses(acc_id):
print('I| done nuking account posts for "{}"'.format(acc_id))
return 0
def logout_account(acc_id):
def pixelfed_logoutall_async():
threading.Thread(target=pixelfed_logoutall).start()
def pixelfed_logoutall():
for acc_id in os.listdir('./db/accounts'):
print('I| logging out account "{}": '.format(acc_id), end='')
if pixelfed_logout(acc_id):
print('ok')
else:
print('not logged')
print('I| done logging out all accounts\n')
def pixelfed_logout(acc_id):
accdata = db_get('accounts', acc_id)
if not pixelfed_islogged(acc_id, accdata):
return False
_, _token = pixelfed_token_url('', accdata['cookie'])
r = requests.post('https://'+config()['instance']+'/logout', data={'_token': _token}, cookies=accdata['cookie'])
del accdata['cookie']
db_set('accounts', acc_id, accdata)
return True
def pixelfed_islogged(acc_id, accdata=None):
if accdata is None:
accdata = db_get('accounts', acc_id)
return 'cookie' in accdata
def pixelfed_loginall_async():
threading.Thread(target=pixelfed_loginall).start()
def pixelfed_loginall():
for acc_id in os.listdir('./db/accounts'):
print('I| logging in account "{}": '.format(acc_id), end='')
if pixelfed_login(acc_id):
print('ok')
else:
print('already logged')
print('I| done logging in all accounts\n')
def pixelfed_login(acc_id, force=False):
# check account is already logged in if not "force"
accdata = db_get('accounts', acc_id)
if not force and pixelfed_islogged(acc_id, accdata):
return
return False
# obtain one time tokens for the pixelfed instance
_cookies, _token = pixelfed_token_url()
@ -165,6 +193,14 @@ def pixelfed_login(acc_id, force=False):
# add the raw cookies to the account data for later calls
accdata['cookie'] = dict(r.cookies)
db_set('accounts', acc_id, accdata)
return True
def pixelfed_islogged(acc_id, accdata=None):
if accdata is None:
accdata = db_get('accounts', acc_id)
return 'cookie' in accdata
def pixelfed_token_url(url='', _cookies=None):
r = requests.get( 'https://'+config()['instance']+url, cookies=_cookies )

75
server.py
View File

@ -6,48 +6,81 @@ import re
class MyServer(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "application/json")
self.end_headers()
# the account is the first part of the URL
# and the action for the account is the last part
# example: /shakira/add
path = self.path.strip('/')
parts = path.split('/')
if len(parts) == 1:
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
# list accounts (plain text)
if parts[0] == 'list':
accounts = os.listdir('./db/accounts')
for acc in sorted(set(accounts)):
self.wfile.write(bytes(acc+'\n', "utf-8"))
elif len(parts) == 2:
# lists accounts on a pretty HTML + CSS
# making sure there is no XSS possible on account names
elif parts[0] == 'mirrors':
self.wfile.write(bytes('HTML', "utf-8"))
return
if len(parts) == 2:
self.send_response(200)
self.send_header("Content-type", "application/json")
self.end_headers()
# a wilcard select all accounts
if parts[0] == '*':
if parts[1].lower() == 'update':
what = parts[0]
action = parts[1].lower()
if what == '*':
if False:
pass
elif action == 'update':
igmirror.update_allaccounts_async()
elif action == 'login':
igmirror.pixelfed_loginall_async()
elif action == 'logout':
igmirror.pixelfed_logoutall_async()
else:
# make sure account name contains only safe characters
# i think IG usernames can only have this characters:
accname = re.sub(r'[^a-zA-Z0-9_\.]+', '', parts[0])
if parts[1].lower() == 'add':
accname = re.sub(r'[^a-zA-Z0-9_\.]+', '', what)
if False:
pass
elif action == 'add':
igmirror.add_igaccount(accname)
elif parts[1].lower() == 'update':
elif action == 'update':
igmirror.update_igaccount_async(accname)
elif parts[1].lower() == 'login':
elif action == 'login':
igmirror.pixelfed_login(accname, True)
elif parts[1].lower() == 'logout':
igmirror.logout_account(accname)
elif parts[1].lower() == 'nuke':
elif action == 'logout':
igmirror.pixelfed_logout(accname)
elif action == 'nuke':
igmirror.delete_statuses(accname)
self.wfile.write(bytes('{"status": "ok"}', "utf-8"))
else:
self.wfile.write(bytes('{"status": "parameters are not correct"}', "utf-8"))
return
self.send_response(400)
self.send_header("Content-type", "application/json")
self.end_headers()
self.wfile.write(bytes('{"status": "parameters are not correct"}', "utf-8"))
if __name__ == "__main__":
addr = '0.0.0.0'
port = 8080
if len(sys.argv) > 1:
port = int(sys.argv[1])
webServer = HTTPServer(('0.0.0.0', port), MyServer)
print("Server started http://%s:%s" % ('0.0.0.0', port))
addr = sys.argv[1]
if len(sys.argv) > 2:
port = sys.argv[2]
webServer = HTTPServer((addr, port), MyServer)
print("Server started http://%s:%s" % (addr, port))
try:
webServer.serve_forever()