1
0
mirror of https://github.com/laurivosandi/certidude synced 2024-12-22 16:25:17 +00:00

Preliminary tests for auth

This commit is contained in:
Lauri Võsandi 2017-04-25 21:47:41 +03:00
parent 3ef4d96b1c
commit 15ae064f55
3 changed files with 52 additions and 1 deletions

View File

@ -1,5 +1,6 @@
import click import click
import falcon
import logging import logging
import os import os
import re import re
@ -133,7 +134,8 @@ def authenticate(optional=False):
import simplepam import simplepam
if not simplepam.authenticate(user, passwd, "sshd"): if not simplepam.authenticate(user, passwd, "sshd"):
logger.critical(u"Basic authentication failed for user %s from %s", logger.critical(u"Basic authentication failed for user %s from %s, "
"are you sure server process has read access to /etc/shadow?",
repr(user), req.context.get("remote_addr")) repr(user), req.context.get("remote_addr"))
raise falcon.HTTPUnauthorized("Forbidden", "Invalid password", ("Basic",)) raise falcon.HTTPUnauthorized("Forbidden", "Invalid password", ("Basic",))

View File

@ -60,6 +60,13 @@ class PosixUserManager(object):
_, _, gid, members = grp.getgrnam(config.ADMIN_GROUP) _, _, gid, members = grp.getgrnam(config.ADMIN_GROUP)
return user.name in members return user.name in members
def all(self):
_, _, gid, members = grp.getgrnam(config.USERS_GROUP)
for username in members:
yield self.get(username)
for user in self.filter_admins(): # TODO: dedup
yield user
class DirectoryConnection(object): class DirectoryConnection(object):
def __enter__(self): def __enter__(self):

View File

@ -1,5 +1,7 @@
import os import os
import requests import requests
import subprocess
import pwd
from falcon import testing from falcon import testing
from click.testing import CliRunner from click.testing import CliRunner
from certidude.cli import entry_point as cli from certidude.cli import entry_point as cli
@ -43,6 +45,39 @@ def test_cli_setup_authority():
assert authority.ca_cert.not_valid_before < datetime.now() assert authority.ca_cert.not_valid_before < datetime.now()
assert authority.ca_cert.not_valid_after > datetime.now() + timedelta(days=7000) assert authority.ca_cert.not_valid_after > datetime.now() + timedelta(days=7000)
try:
pwd.getpwnam("userbot")
except KeyError:
# useradd userbot -G users -p '$1$PBkf5waA$n9EV6WJ7PS6lyGWkgeTPf1'
cmd = "useradd", "userbot", "-G", "users", "-p", "$1$PBkf5waA$n9EV6WJ7PS6lyGWkgeTPf1" # bot
subprocess.call(cmd)
try:
pwd.getpwnam("adminbot")
except KeyError:
# Note: on Fedora use group 'wheel' instead of 'sudo'
# useradd adminbot -G sudo -p '$1$PBkf5waA$n9EV6WJ7PS6lyGWkgeTPf1'
cmd = "useradd", "adminbot", "-G", "sudo", "-p", "$1$PBkf5waA$n9EV6WJ7PS6lyGWkgeTPf1" # bot
subprocess.call(cmd)
usertoken = "Basic dXNlcmJvdDpib3Q="
admintoken = "Basic YWRtaW5ib3Q6Ym90"
result = runner.invoke(cli, ['users'])
assert not result.exception
# Test session API call
r = client().simulate_get("/api/", headers={"Authorization":usertoken})
assert r.status_code == 200
r = client().simulate_get("/api/", headers={"Authorization":admintoken})
assert r.status_code == 200
r = client().simulate_get("/api/")
assert r.status_code == 401
# Try starting up forked server # Try starting up forked server
result = runner.invoke(cli, ['serve', '-f', '-p', '8080']) result = runner.invoke(cli, ['serve', '-f', '-p', '8080'])
assert not result.exception assert not result.exception
@ -172,6 +207,13 @@ def test_cli_setup_authority():
r = client().simulate_get("/api/signed/test2/tag/") r = client().simulate_get("/api/signed/test2/tag/")
assert r.status_code == 401 assert r.status_code == 401
r = client().simulate_get("/api/signed/test2/tag/", headers={"Authorization":usertoken})
assert r.status_code == 403
r = client().simulate_get("/api/signed/test2/tag/", headers={"Authorization":admintoken})
assert r.status_code == 200
# Revoke all valid ones # Revoke all valid ones
result = runner.invoke(cli, ['revoke', 'test2']) result = runner.invoke(cli, ['revoke', 'test2'])
assert not result.exception assert not result.exception