Recently Written · git

sh

Small shell and Python scripts

git clone https://github.com/equwal/sh

Log | Files | Refs


http-collect (4652 bytes)

1 #!/usr/bin/env python3
2 # Serve a secret URL (and QR code) where a phone uploads files/text to cwd.
3 # Usage: http-collect [--host IP] [--port N] [--secret S]  Needs: flask, qrcode
4 
5 
6 from argparse import ArgumentParser
7 from os.path import lexists
8 from random import randint
9 from re import match
10 from socket import socket, AF_INET, SOCK_STREAM
11 from subprocess import check_output
12 from sys import stderr, stdout
13 from time import sleep
14 from uuid import uuid4
15 
16 from flask import Flask, request
17 from qrcode import QRCode
18 
19 
20 app = Flask(__name__)
21 secret = str(uuid4())
22 
23 
24 def html(body):
25     out = '''
26 <!DOCTYPE html>
27 <html>
28     <head>
29         <link href="data:," rel="icon">
30         <meta charset="UTF-8">
31         <meta name="viewport" content="width=device-width,initial-scale=1.0">
32         <title>http-collect</title>
33         <style>
34 body {
35     max-width: 400px;
36     margin: 0px auto;
37     padding: 3em;
38     font-family: sans-serif;
39 }
40 input, textarea {
41     width: 100%;
42     height: 8em;
43     margin: 0.5em 0px;
44 }
45         </style>
46     </head>
47     <body>
48     '''
49     out += body
50     out += '''
51     </body>
52 </html>
53     '''
54 
55     return out
56 
57 
58 def sanitize_fs(s):
59     return s.strip().lstrip('.').replace('/', '_') or 'unnamed'
60 
61 
62 def sanitize_html(s):
63     return s.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
64 
65 
66 def suffixify(basename, suff):
67     m = match(r'(.+)\.([^.]+)$', basename)
68     if not m:
69         return f'{basename} ({suff:03d})'
70 
71     return f'{m.group(1)} ({suff:03d}).{m.group(2)}'
72 
73 
74 def get_final_name(original_name):
75     candidate = base_name = sanitize_fs(original_name)
76     suff = 1
77     while lexists(candidate):
78         candidate = suffixify(base_name, suff)
79         suff += 1
80     return candidate
81 
82 
83 def determine_host_and_port(args):
84     host = None
85     port = None
86 
87     if args.host:
88         host = args.host
89     if args.port:
90         port = args.port
91 
92     if host is None:
93         try:
94             rt = check_output(['ip', '-4', 'route']).decode('UTF-8')
95         except OSError:
96             rt = ''
97         for line in rt.splitlines():
98             m = match(r'default.* src ([0-9.]+)', line)
99             if m:
100                 host = m.group(1)
101 
102         if host is None:
103             print('Could not find my IPv4, using 0.0.0.0', file=stderr)
104             host = '0.0.0.0'
105 
106     success = False
107     for _ in range(5):
108         if port is None:
109             try_port = randint(1025, 65535)
110         else:
111             try_port = port
112 
113         try:
114             with socket(AF_INET, SOCK_STREAM) as s:
115                 s.bind((host, try_port))
116                 s.listen()
117         except Exception as e:
118             print(f'Failed to use {host}:{try_port}: {repr(e)}')
119             sleep(1)
120         else:
121             port = try_port
122             success = True
123             break
124 
125     if not success:
126         raise Exception('Could not find free port')
127 
128     url = f'http://{host}:{port}/{secret}'
129 
130     qr = QRCode()
131     qr.add_data(f'http://{host}:{port}/{secret}')
132     qr.print_ascii(tty=stdout.isatty())
133 
134     print()
135     print(url)
136     print()
137 
138     return host, port
139 
140 
141 @app.route('/<secret_from_url>', methods=['GET'])
142 def form(secret_from_url):
143     if secret_from_url != secret:
144         return 'no', 403
145 
146     return html('''
147     <form method="POST" enctype="multipart/form-data">
148         <input type="file" name="file" multiple>
149         <textarea name="textarea"></textarea>
150         <input type="submit">
151     </form>
152     ''')
153 
154 
155 @app.route('/<secret_from_url>', methods=['POST'])
156 def save(secret_from_url):
157     if secret_from_url != secret:
158         return 'no', 403
159 
160     saved = []
161 
162     for file in request.files.getlist('file'):
163         if file.filename:
164             final_name = get_final_name(file.filename)
165             file.save(final_name)
166             saved.append(f'<li>{sanitize_html(final_name)}</li>')
167 
168     if request.form.get('textarea'):
169         final_name = get_final_name('plain.txt')
170         with open(final_name, 'w', encoding='UTF-8') as fp:
171             fp.write(request.form.get('textarea'))
172         saved.append(f'<li>{sanitize_html(final_name)}</li>')
173 
174     num = len(saved)
175     suffix = '' if num == 1 else 's'
176     body = f'<p>{num} file{suffix} saved:</p><ul>'
177     body += '\n'.join(saved)
178     body += '</ul>'
179 
180     body += f'<p><a href="/{secret}">Back</a></p>'
181     return html(body)
182 
183 
184 if __name__ == '__main__':
185     parser = ArgumentParser()
186     parser.add_argument('--host')
187     parser.add_argument('--port', type=int)
188     parser.add_argument('--secret')
189     args = parser.parse_args()
190 
191     if args.secret:
192         secret = args.secret
193 
194     host, port = determine_host_and_port(args)
195     app.run(host=host, port=port)