|
| 1 | +#!/usr/bin/env python |
| 2 | + |
| 3 | +import pathlib as pl, contextlib as cl, hashlib as hl |
| 4 | +import os, sys, re, random, stat, tempfile, fcntl, base64 |
| 5 | + |
| 6 | + |
| 7 | +p_out = lambda *a,**kw: print(*a, **kw, flush=True) |
| 8 | +p_err = lambda *a,**kw: print(*a, **kw, file=sys.stderr, flush=True) or 1 |
| 9 | + |
| 10 | +class adict(dict): |
| 11 | + def __init__(self, *args, **kws): |
| 12 | + super().__init__(*args, **kws) |
| 13 | + self.__dict__ = self |
| 14 | + |
| 15 | +@cl.contextmanager |
| 16 | +def safe_replacement( path, *open_args, |
| 17 | + mode=None, commit=True, xattrs=None, uidgid=None, **open_kws ): |
| 18 | + 'Context to atomically create/replace file-path in-place unless errors are raised' |
| 19 | + path = str(path); exists = st = None |
| 20 | + if mode is None: |
| 21 | + try: mode = stat.S_IMODE((st := os.lstat(path)).st_mode) |
| 22 | + except FileNotFoundError: exists = False |
| 23 | + if uidgid is False or exists is False or isinstance(uidgid, tuple): pass |
| 24 | + elif uidgid is True or os.geteuid(): |
| 25 | + try: st = st or os.stat(path); uidgid = st.st_uid, st.st_gid |
| 26 | + except FileNotFoundError: pass |
| 27 | + if xattrs is None and getattr(os, 'getxattr', None): # MacOS |
| 28 | + try: xattrs = dict((k, os.getxattr(path, k)) for k in os.listxattr(path)) |
| 29 | + except FileNotFoundError: exists = False |
| 30 | + except OSError as err: |
| 31 | + if err.errno != errno.ENOTSUP: raise |
| 32 | + open_kws.update( delete=False, |
| 33 | + dir=os.path.dirname(path), prefix=os.path.basename(path) + '.' ) |
| 34 | + if not open_args: open_kws.setdefault('mode', 'w') |
| 35 | + if exists is None: exists = os.path.exists(path) |
| 36 | + with tempfile.NamedTemporaryFile(*open_args, **open_kws) as tmp: |
| 37 | + try: |
| 38 | + if mode is not None: os.fchmod(tmp.fileno(), mode) |
| 39 | + if xattrs: |
| 40 | + for k, v in xattrs.items(): os.setxattr(tmp.name, k, v) |
| 41 | + if isinstance(uidgid, tuple): os.fchown(tmp.fileno(), *uidgid) |
| 42 | + tmp.file_path, tmp.file_exists, tmp.commit = path, exists, not commit |
| 43 | + yield tmp |
| 44 | + if not tmp.closed: tmp.flush() |
| 45 | + try: os.fdatasync(tmp) |
| 46 | + except AttributeError: pass # MacOS |
| 47 | + if tmp.commit: os.rename(tmp.name, path) |
| 48 | + finally: |
| 49 | + try: os.unlink(tmp.name) |
| 50 | + except FileNotFoundError: pass |
| 51 | + |
| 52 | +def gen_new_auth(src_auth, fmt, max_creds, retries=10_000): |
| 53 | + lines, logins = list(), dict() |
| 54 | + for n, line in enumerate(src_auth): |
| 55 | + lines.append(line.rstrip()) |
| 56 | + if m := re.match(f'({fmt.login_re}):', line): logins.setdefault(m[1], list()).append(n) |
| 57 | + while (login := fmt.login.format(**fmt.vars_gen())) in logins: |
| 58 | + if (retries := retries - 1) <= 0: raise OverflowError('Failed to generate unique login') |
| 59 | + lines_rm, logins[login], pw = set(), len(lines), fmt.pw.format(**fmt.vars_gen()) |
| 60 | + pw_ssha = pw.encode() + (salt := os.urandom(4)) |
| 61 | + pw_ssha = base64.b64encode(hl.sha1(pw_ssha).digest() + salt).decode() |
| 62 | + lines.append(login + ':{SSHA}' + pw_ssha) |
| 63 | + while len(logins) > max_creds: |
| 64 | + login_old = list(logins)[0]; lines_rm.update(logins.pop(login_old)) |
| 65 | + for n in sorted(lines_rm, reverse=True): del lines[n] |
| 66 | + return login, pw, lines |
| 67 | + |
| 68 | +def main(args=None): |
| 69 | + import argparse, textwrap |
| 70 | + dd = lambda text: re.sub( r' \t+', ' ', |
| 71 | + textwrap.dedent(text).strip('\n') + '\n' ).replace('\t', ' ') |
| 72 | + parser = argparse.ArgumentParser( |
| 73 | + formatter_class=argparse.RawTextHelpFormatter, description=dd(''' |
| 74 | + Generate/replace password in nginx auth_basic_user_file and static html page(s). |
| 75 | + Keeps some number of old credentials in the auth-file, |
| 76 | + uses simplest/weak {SSHA} password scheme there, puts current/latest token into html. |
| 77 | + Intended be used for weak/temp auto-rotating http auth, presented on some page.''')) |
| 78 | + parser.add_argument('auth_file', help=dd(''' |
| 79 | + Path to nginx auth_basic_user_file, where credentials are rotated. |
| 80 | + Auth lines are added at the end there, ones matching old login names |
| 81 | + are removed from the top, when number of matches ones goes over specified limit.''')) |
| 82 | + parser.add_argument('-l', '--auth-login', metavar='fmt', default='hooman{nnn}', help=dd(''' |
| 83 | + Python string-format for logins that are managed by this script. |
| 84 | + Should contain "n" template-variables, which will be substituted-with/match |
| 85 | + random number with that number of digits (e.g. 3 for {nnn}). Default: %(default)s''')) |
| 86 | + parser.add_argument('-p', '--auth-pw', metavar='fmt', default='token{nnnn}', help=dd(''' |
| 87 | + Python string-format for passwords generated by this script. |
| 88 | + Same idea as with -l/--auth-login. Default: %(default)s''')) |
| 89 | + parser.add_argument('-n', '--auth-lines', metavar='N', default=3, help=dd(''' |
| 90 | + Total number of login:pw credential lines |
| 91 | + matching -l/--auth-login format to keep in auth_file. |
| 92 | + Lines are added at the end there, ones matching old login names |
| 93 | + are removed from the top, when number them goes over this limit. Default: %(default)s |
| 94 | + Login gets re-generated until it doesn't match any pre-existing line(s) there.''')) |
| 95 | + parser.add_argument('-f', '--html-file', metavar='file', action='append', help=dd(''' |
| 96 | + HTML file(s) where all -l/--auth-login and -p/--auth-pw |
| 97 | + patterns are regexp-matched and replaced with latest generated credentials. |
| 98 | + Matching does not look for old/removed creds specifically, just anything |
| 99 | + that looks exactly like login/pw, so make sure to use something unique there. |
| 100 | + At least one replacement for login/pw must be made in every |
| 101 | + specified HTML file, otherwise error is raised instead of changing anything. |
| 102 | + Can be used multiple times to process any number of such files.''')) |
| 103 | + parser.add_argument('-v', '--verbose', action='store_true', help=dd(''' |
| 104 | + Print new credentials, new nginx auth file, |
| 105 | + and number of replacements made in HTML file(s).''')) |
| 106 | + parser.add_argument('--dry-run', action='store_true', help=dd(''' |
| 107 | + Same as -v/--verbose, but also don't change any actual files, only print info.''')) |
| 108 | + opts = parser.parse_args(sys.argv[1:] if args is None else args) |
| 109 | + |
| 110 | + nn_max, fmt = 20, adict(login=opts.auth_login, pw=opts.auth_pw) |
| 111 | + fmt.vars_gen = lambda: dict( |
| 112 | + ('n'*n, f'{{:0{n}d}}'.format(random.randint(0, 10**n-1))) for n in range(1, nn_max) ) |
| 113 | + vars_re = dict(('n'*n, r'\d'*n) for n in range(1, nn_max)) |
| 114 | + fmt.login_re, fmt.pw_re = (fmt.format(**vars_re) for fmt in [fmt.login, fmt.pw]) |
| 115 | + p_auth, p_htmls = pl.Path(opts.auth_file), list(map(pl.Path, opts.html_file or list())) |
| 116 | + |
| 117 | + with cl.ExitStack() as ctx: |
| 118 | + src_auth = (cte := ctx.enter_context)(p_auth.open('r+')) |
| 119 | + fcntl.lockf(src_auth, fcntl.LOCK_EX) |
| 120 | + login, pw, auth_lines = gen_new_auth(src_auth, fmt, opts.auth_lines) |
| 121 | + |
| 122 | + for n, p in enumerate(p_htmls): |
| 123 | + p_htmls[n] = [p, src := cte(p.open('r+')), cte(safe_replacement(p))] |
| 124 | + fcntl.lockf(src, fcntl.LOCK_EX) |
| 125 | + dst_auth = cte(safe_replacement(p_auth)) |
| 126 | + for n, (p, src, dst) in enumerate(p_htmls): |
| 127 | + html = src.read() |
| 128 | + for rxn, (new, fmt, rx) in enumerate([ |
| 129 | + (login, fmt.login, fmt.login_re), (pw, fmt.pw, fmt.pw_re) ], 1): |
| 130 | + html, rn = re.subn(fr'\b{rx}\b', new, html); p_htmls[n][rxn] = rn |
| 131 | + if rn: continue |
| 132 | + return p_err(f'ERROR: HTML file has no pattern-matches for {fmt!r} [ {p} ]') |
| 133 | + dst.write(html); dst.commit = not opts.dry_run |
| 134 | + dst_auth.write(''.join(f'{line}\n' for line in auth_lines)) |
| 135 | + dst_auth.commit = not opts.dry_run |
| 136 | + |
| 137 | + if opts.verbose or opts.dry_run: |
| 138 | + p_out(f'Generated new login=[ {login} ] pw=[ {pw} ]') |
| 139 | + p_out(f'nginx auth file lines [ {p_auth} ]:') |
| 140 | + for line in auth_lines: p_out(f' {line}') |
| 141 | + if p_htmls: |
| 142 | + p_out('Replacements in HTML file(s):') |
| 143 | + for p, login_rn, pw_rn in p_htmls: p_out(f' {p} [ login={login_rn} pw={pw_rn} ]') |
| 144 | + |
| 145 | +if __name__ == '__main__': sys.exit(main()) |
0 commit comments