mirror of
https://github.com/ceph/ceph-csi.git
synced 2024-11-27 00:30:18 +00:00
154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
#
|
|
# Copyright (C) 2008-2013 Edgewall Software
|
|
# Copyright (C) 2008 Eli Carter
|
|
# All rights reserved.
|
|
#
|
|
# This software is licensed as described in the file COPYING, which
|
|
# you should have received as part of this distribution. The terms
|
|
# are also available at http://trac.edgewall.com/license.html.
|
|
#
|
|
# This software consists of voluntary contributions made by many
|
|
# individuals. For the exact contribution history, see the revision
|
|
# history and logs, available at http://trac.edgewall.org/.
|
|
|
|
"""Replacement for htpasswd"""
|
|
|
|
import os
|
|
import sys
|
|
import random
|
|
from optparse import OptionParser
|
|
|
|
# We need a crypt module, but Windows doesn't have one by default. Try to find
|
|
# one, and tell the user if we can't.
|
|
try:
|
|
import crypt
|
|
except ImportError:
|
|
try:
|
|
import fcrypt as crypt
|
|
except ImportError:
|
|
sys.stderr.write("Cannot find a crypt module. "
|
|
"Possibly http://carey.geek.nz/code/python-fcrypt/\n")
|
|
sys.exit(1)
|
|
|
|
|
|
def wait_for_file_mtime_change(filename):
|
|
"""This function is typically called before a file save operation,
|
|
waiting if necessary for the file modification time to change. The
|
|
purpose is to avoid successive file updates going undetected by the
|
|
caching mechanism that depends on a change in the file modification
|
|
time to know when the file should be reparsed."""
|
|
try:
|
|
mtime = os.stat(filename).st_mtime
|
|
os.utime(filename, None)
|
|
while mtime == os.stat(filename).st_mtime:
|
|
time.sleep(1e-3)
|
|
os.utime(filename, None)
|
|
except OSError:
|
|
pass # file doesn't exist (yet)
|
|
|
|
def salt():
|
|
"""Returns a string of 2 randome letters"""
|
|
letters = 'abcdefghijklmnopqrstuvwxyz' \
|
|
'ABCDEFGHIJKLMNOPQRSTUVWXYZ' \
|
|
'0123456789/.'
|
|
return random.choice(letters) + random.choice(letters)
|
|
|
|
|
|
class HtpasswdFile:
|
|
"""A class for manipulating htpasswd files."""
|
|
|
|
def __init__(self, filename, create=False):
|
|
self.entries = []
|
|
self.filename = filename
|
|
if not create:
|
|
if os.path.exists(self.filename):
|
|
self.load()
|
|
else:
|
|
raise Exception("%s does not exist" % self.filename)
|
|
|
|
def load(self):
|
|
"""Read the htpasswd file into memory."""
|
|
lines = open(self.filename, 'r').readlines()
|
|
self.entries = []
|
|
for line in lines:
|
|
username, pwhash = line.split(':')
|
|
entry = [username, pwhash.rstrip()]
|
|
self.entries.append(entry)
|
|
|
|
def save(self):
|
|
"""Write the htpasswd file to disk"""
|
|
wait_for_file_mtime_change(self.filename)
|
|
open(self.filename, 'w').writelines(["%s:%s\n" % (entry[0], entry[1])
|
|
for entry in self.entries])
|
|
|
|
def update(self, username, password):
|
|
"""Replace the entry for the given user, or add it if new."""
|
|
pwhash = crypt.crypt(password, salt())
|
|
matching_entries = [entry for entry in self.entries
|
|
if entry[0] == username]
|
|
if matching_entries:
|
|
matching_entries[0][1] = pwhash
|
|
else:
|
|
self.entries.append([username, pwhash])
|
|
|
|
def delete(self, username):
|
|
"""Remove the entry for the given user."""
|
|
self.entries = [entry for entry in self.entries
|
|
if entry[0] != username]
|
|
|
|
|
|
def main():
|
|
"""
|
|
%prog -b[c] filename username password
|
|
%prog -D filename username"""
|
|
# For now, we only care about the use cases that affect tests/functional.py
|
|
parser = OptionParser(usage=main.__doc__)
|
|
parser.add_option('-b', action='store_true', dest='batch', default=False,
|
|
help='Batch mode; password is passed on the command line IN THE CLEAR.'
|
|
)
|
|
parser.add_option('-c', action='store_true', dest='create', default=False,
|
|
help='Create a new htpasswd file, overwriting any existing file.')
|
|
parser.add_option('-D', action='store_true', dest='delete_user',
|
|
default=False, help='Remove the given user from the password file.')
|
|
|
|
options, args = parser.parse_args()
|
|
|
|
def syntax_error(msg):
|
|
"""Utility function for displaying fatal error messages with usage
|
|
help.
|
|
"""
|
|
sys.stderr.write("Syntax error: " + msg)
|
|
sys.stderr.write(parser.get_usage())
|
|
sys.exit(1)
|
|
|
|
if not (options.batch or options.delete_user):
|
|
syntax_error("Only batch and delete modes are supported\n")
|
|
|
|
# Non-option arguments
|
|
if len(args) < 2:
|
|
syntax_error("Insufficient number of arguments.\n")
|
|
filename, username = args[:2]
|
|
if options.delete_user:
|
|
if len(args) != 2:
|
|
syntax_error("Incorrect number of arguments.\n")
|
|
password = None
|
|
else:
|
|
if len(args) != 3:
|
|
syntax_error("Incorrect number of arguments.\n")
|
|
password = args[2]
|
|
|
|
passwdfile = HtpasswdFile(filename, create=options.create)
|
|
|
|
if options.delete_user:
|
|
passwdfile.delete(username)
|
|
else:
|
|
passwdfile.update(username, password)
|
|
|
|
passwdfile.save()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|