commit 61374e95bc4a0a563198d48215fedd3c337e4957
parent cc8406cdbbe4a35a08fb72462c343623ffa8bb5f
Author: Asher Morgan <asher@ashermorgan.net>
Date: Sat, 29 Aug 2026 08:00:26 -0700
Implement playbook password protection
Diffstat:
6 files changed, 137 insertions(+), 105 deletions(-)
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,6 @@
.env
playbooks.yml
+passwords.yml
*.pyc
__pycache__
.venv
diff --git a/aoh/client.py b/aoh/client.py
@@ -14,6 +14,7 @@ from urllib.request import Request, urlopen
API = '{{ API_URL }}' # Substitution performed by Flask
PASSWORD_PROMPTS = {
+ 'aoh_password': 'AoH password: ',
'connection_password': 'SSH password: ',
'become_password': 'BECOME password: ',
'vault_password': 'Vault password: ',
diff --git a/aoh/playbook.py b/aoh/playbook.py
@@ -1,69 +1,10 @@
import os
-import tempfile
-import ansible_runner
import yaml
PLAYBOOKS_FILE = os.getenv('AOH_PLAYBOOKS_FILE', 'playbooks.yml')
PLAYBOOKS_DIR = os.getenv('AOH_PLAYBOOKS_DIR', os.path.dirname(PLAYBOOKS_FILE))
-_CLI_OPT_WHITELIST = [
- # These options should be safe for clients to invoke
- '--ask-become-pass', '-K',
- '--ask-pass', '-k',
- '--ask-vault-password', '--ask-vault-pass', '-J',
- '--check', '-C',
- '--diff', '-D',
- '--extra-vars', '-e',
- '--help', '-h',
- '--limit', '-l',
- '--list-tags',
- '--skip-tags',
- '--start-at-task',
- '--tags', '-t',
- '--verbose', '-v',
- '--version',
-]
-
-_CLI_OPT_BLACKLIST = [
- # These options are not supported at all by AoH
- '--step', # Interactive
-]
-
-_CLI_PASSWORD_OPTS = {
- # These command line flags require us to prompt the user for a password
- '--ask-become-pass': 'become_password',
- '--ask-pass': 'connection_password',
- '--ask-vault-pass': 'vault_password',
- '--ask-vault-password': 'vault_password',
- '-J': 'vault_password',
- '-K': 'become_password',
- '-k': 'connection_password',
-}
-
-_CONFIG_PASSWORD_OPTS = {
- # These config options require us to prompt the user for a password
- 'DEFAULT_ASK_PASS': 'connection_password',
- 'DEFAULT_ASK_VAULT_PASS': 'vault_password',
- 'DEFAULT_BECOME_ASK_PASS': 'become_password',
-}
-
-
-def _get_opts(args):
- """Identify options present in a list of CLI arguments."""
-
- opts = []
-
- for arg in args:
- if arg == '--':
- break
- elif arg.startswith('--'):
- opts += [arg.split('=')[0]]
- elif arg.startswith('-'):
- for opt in arg.split('=')[0][1:]:
- opts += ['-' + opt]
-
- return opts
class PlaybookError(Exception):
@@ -77,6 +18,7 @@ class Playbook:
_types = {
'path': str,
'config': str,
+ 'password': bool,
'cmdline': str,
'output': bool,
'jinja': bool,
@@ -88,8 +30,10 @@ class Playbook:
if key in dict and type(dict[key]) != _type:
raise PlaybookError(f'"{key}" field must be of type {_type}')
+ self.name = name
self.path = dict.get('path', name)
self.config = dict.get('config', None)
+ self.password = dict.get('password', False)
self.cmdline = dict.get('cmdline', '')
self.output = dict.get('output', True)
self.jinja = dict.get('jinja', False)
@@ -100,50 +44,6 @@ class Playbook:
self.config = os.path.abspath(os.path.join(PLAYBOOKS_DIR, self.config))
- def validate_args(self, args):
- """Determine if arguments pass the playbook's security policies."""
-
- cmdline = ' '.join(args)
- if not self.jinja and ('{{' in cmdline or '{%' in cmdline):
- return False
-
- for opt in _get_opts(args):
- if (not (opt in _CLI_OPT_WHITELIST or opt in self.allow_opts) or
- (opt in _CLI_OPT_BLACKLIST or opt in self.block_opts)):
- return False
-
- return True
-
-
- def get_required_passwords(self, args):
- """Determine what passwords the user must be prompted for."""
-
- pw_types = set()
-
- opts = _get_opts(args)
- for opt, pw_type in _CLI_PASSWORD_OPTS.items():
- if opt in opts:
- pw_types.add(pw_type)
-
- with tempfile.TemporaryDirectory(prefix='aoh-') as dir:
- raw_config = ansible_runner.get_ansible_config(
- 'dump',
- self.config,
- private_data_dir=dir,
- quiet=True,
- )[0]
-
- for opt, pw_type in _CONFIG_PASSWORD_OPTS.items():
- val = eval(next(
- x for x in raw_config.split('\n')
- if x.startswith(opt)
- ).split('= ')[1])
- if val:
- pw_types.add(pw_type)
-
- return list(pw_types)
-
-
def get_playbook(name):
"""Lookup a playbook configuration."""
diff --git a/aoh/security.py b/aoh/security.py
@@ -0,0 +1,124 @@
+import os
+import tempfile
+
+import ansible_runner
+import bcrypt
+import yaml
+
+_PASSWORDS_FILE = os.getenv('AOH_PASSWORDS_FILE', 'passwords.yml')
+
+_CLI_OPT_WHITELIST = [
+ # These options should be safe for clients to invoke
+ '--ask-become-pass', '-K',
+ '--ask-pass', '-k',
+ '--ask-vault-password', '--ask-vault-pass', '-J',
+ '--check', '-C',
+ '--diff', '-D',
+ '--extra-vars', '-e',
+ '--help', '-h',
+ '--limit', '-l',
+ '--list-tags',
+ '--skip-tags',
+ '--start-at-task',
+ '--tags', '-t',
+ '--verbose', '-v',
+ '--version',
+]
+
+_CLI_OPT_BLACKLIST = [
+ # These options are not supported at all by AoH
+ '--step', # Interactive
+]
+
+_CLI_PASSWORD_OPTS = {
+ # These command line flags require us to prompt the user for a password
+ '--ask-become-pass': 'become_password',
+ '--ask-pass': 'connection_password',
+ '--ask-vault-pass': 'vault_password',
+ '--ask-vault-password': 'vault_password',
+ '-J': 'vault_password',
+ '-K': 'become_password',
+ '-k': 'connection_password',
+}
+
+_CONFIG_PASSWORD_OPTS = {
+ # These config options require us to prompt the user for a password
+ 'DEFAULT_ASK_PASS': 'connection_password',
+ 'DEFAULT_ASK_VAULT_PASS': 'vault_password',
+ 'DEFAULT_BECOME_ASK_PASS': 'become_password',
+}
+
+
+def _get_opts(args):
+ """Identify options present in a list of CLI arguments."""
+
+ opts = []
+
+ for arg in args:
+ if arg.startswith('--'):
+ opts += [arg.split('=')[0]]
+ elif arg.startswith('-'):
+ for opt in arg.split('=')[0][1:]:
+ opts += ['-' + opt]
+
+ return opts
+
+
+def validate_args(playbook, args):
+ """Determine if arguments pass a playbook's security policies."""
+
+ cmdline = ' '.join(args)
+ if not playbook.jinja and ('{{' in cmdline or '{%' in cmdline):
+ return False
+
+ for opt in _get_opts(args):
+ if (not (opt in _CLI_OPT_WHITELIST or opt in playbook.allow_opts) or
+ (opt in _CLI_OPT_BLACKLIST or opt in playbook.block_opts)):
+ return False
+
+ return True
+
+
+def get_required_passwords(playbook, args):
+ """Determine what passwords the user must be prompted for."""
+
+ pw_types = set()
+
+ if playbook.password:
+ pw_types.add('aoh_password')
+
+ opts = _get_opts(args)
+ for opt, pw_type in _CLI_PASSWORD_OPTS.items():
+ if opt in opts:
+ pw_types.add(pw_type)
+
+ with tempfile.TemporaryDirectory(prefix='aoh-') as dir:
+ raw_config = ansible_runner.get_ansible_config(
+ 'dump',
+ playbook.config,
+ private_data_dir=dir,
+ quiet=True,
+ )[0]
+
+ for opt, pw_type in _CONFIG_PASSWORD_OPTS.items():
+ val = eval(next(
+ x for x in raw_config.split('\n')
+ if x.startswith(opt)
+ ).split('= ')[1])
+ if val:
+ pw_types.add(pw_type)
+
+ return list(pw_types)
+
+
+def validate_aoh_password(playbook, password):
+ """Determine whether a provided AoH password is valid for a playbook."""
+
+ if not playbook.password:
+ return True
+ if not password:
+ return False
+
+ with open(_PASSWORDS_FILE, 'r') as f:
+ hash = yaml.safe_load(f).get(playbook.name)
+ return bcrypt.checkpw(password.encode(), hash.encode())
diff --git a/aoh/server.py b/aoh/server.py
@@ -5,6 +5,7 @@ from flask_apscheduler import APScheduler
from .playbook import get_playbook
from .runner import Runner
+from .security import *
_RUNNERS = {}
_GC_INTERVAL = 60 # 1 minute
@@ -40,10 +41,10 @@ def new_runner():
if not playbook:
return { 'err': f"Playbook not found: {request.json['playbook']}" }, 400
- if not playbook.validate_args(request.json['args']):
+ if not validate_args(playbook, request.json['args']):
return { 'err': 'Bad or banned arguments passed.' }, 400
- exp_pw_types = playbook.get_required_passwords(request.json['args'])
+ exp_pw_types = get_required_passwords(playbook, request.json['args'])
act_pws = request.json.get('passwords', {})
missing_pw_types = [pw_type for pw_type in exp_pw_types if pw_type not in
act_pws]
@@ -53,6 +54,10 @@ def new_runner():
'passwords': exp_pw_types,
}, 401
+ if not validate_aoh_password(playbook, act_pws.get('aoh_password')):
+ return { 'err': 'Incorrect AoH password.' }, 401
+ act_pws.pop('aoh_password', None) # Don't pass AoH password on to runner
+
runner = Runner(playbook, request.json['host'], request.json['args'],
act_pws)
diff --git a/requirements.txt b/requirements.txt
@@ -5,3 +5,4 @@ flask-apscheduler
gunicorn
python-dotenv
pyyaml
+bcrypt
