mirror of
				https://github.com/k3s-io/kubernetes.git
				synced 2025-10-31 05:40:42 +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()
 |