diff --git a/lemur/acme_providers/cli.py b/lemur/acme_providers/cli.py index 310efad1..7efa196e 100644 --- a/lemur/acme_providers/cli.py +++ b/lemur/acme_providers/cli.py @@ -1,12 +1,15 @@ import time import json +import arrow from flask_script import Manager from flask import current_app from lemur.extensions import sentry from lemur.constants import SUCCESS_METRIC_STATUS +from lemur.plugins import plugins from lemur.plugins.lemur_acme.plugin import AcmeHandler +from lemur.plugins.lemur_aws import s3 manager = Manager( usage="Handles all ACME related tasks" @@ -84,3 +87,105 @@ def dnstest(domain, token): status = SUCCESS_METRIC_STATUS print("[+] Done with ACME Tests.") + + +@manager.option( + "-t", + "--token", + dest="token", + default="date: " + arrow.utcnow().format("YYYY-MM-DDTHH-mm-ss"), + required=False, + help="Value of the Token", +) +@manager.option( + "-n", + "--token_name", + dest="token_name", + default="Token-" + arrow.utcnow().format("YYYY-MM-DDTHH-mm-ss"), + required=False, + help="path", +) +@manager.option( + "-p", + "--prefix", + dest="prefix", + default="test/", + required=False, + help="S3 bucket prefix", +) +@manager.option( + "-a", + "--account_number", + dest="account_number", + required=True, + help="AWS Account", +) +@manager.option( + "-b", + "--bucket_name", + dest="bucket_name", + required=True, + help="Bucket Name", +) +def upload_acme_token_s3(token, token_name, prefix, account_number, bucket_name): + """ + This method serves for testing the upload_acme_token to S3, fetching the token to verify it, and then deleting it. + It mainly serves for testing purposes. + :param token: + :param token_name: + :param prefix: + :param account_number: + :param bucket_name: + :return: + """ + additional_options = [ + { + "name": "bucket", + "value": bucket_name, + "type": "str", + "required": True, + "validation": r"[0-9a-z.-]{3,63}", + "helpMessage": "Must be a valid S3 bucket name!", + }, + { + "name": "accountNumber", + "type": "str", + "value": account_number, + "required": True, + "validation": r"[0-9]{12}", + "helpMessage": "A valid AWS account number with permission to access S3", + }, + { + "name": "region", + "type": "str", + "default": "us-east-1", + "required": False, + "helpMessage": "Region bucket exists", + "available": ["us-east-1", "us-west-2", "eu-west-1"], + }, + { + "name": "encrypt", + "type": "bool", + "value": False, + "required": False, + "helpMessage": "Enable server side encryption", + "default": True, + }, + { + "name": "prefix", + "type": "str", + "value": prefix, + "required": False, + "helpMessage": "Must be a valid S3 object prefix!", + }, + ] + + p = plugins.get("aws-s3") + p.upload_acme_token(token_name, token, additional_options) + + if not prefix.endswith("/"): + prefix + "/" + + token_res = s3.get(bucket_name, prefix + token_name, account_number=account_number) + assert(token_res == token) + s3.delete(bucket_name, prefix + token_name, account_number=account_number) diff --git a/lemur/auth/ldap.py b/lemur/auth/ldap.py index ed87b76c..030c7c78 100644 --- a/lemur/auth/ldap.py +++ b/lemur/auth/ldap.py @@ -210,7 +210,8 @@ class LdapPrincipal: self.ldap_groups = [] for group in lgroups: (dn, values) = group - self.ldap_groups.append(values["cn"][0].decode("ascii")) + if type(values) == dict: + self.ldap_groups.append(values["cn"][0].decode("ascii")) else: lgroups = self.ldap_client.search_s( self.ldap_base_dn, ldap.SCOPE_SUBTREE, ldap_filter, self.ldap_attrs diff --git a/lemur/auth/service.py b/lemur/auth/service.py index 0e1521b3..f954ce51 100644 --- a/lemur/auth/service.py +++ b/lemur/auth/service.py @@ -101,7 +101,8 @@ def login_required(f): return dict(message="Token is invalid"), 403 try: - payload = jwt.decode(token, current_app.config["LEMUR_TOKEN_SECRET"]) + header_data = fetch_token_header(token) + payload = jwt.decode(token, current_app.config["LEMUR_TOKEN_SECRET"], algorithms=[header_data["alg"]]) except jwt.DecodeError: return dict(message="Token is invalid"), 403 except jwt.ExpiredSignatureError: diff --git a/lemur/authorities/models.py b/lemur/authorities/models.py index 61ab779e..94985cc9 100644 --- a/lemur/authorities/models.py +++ b/lemur/authorities/models.py @@ -18,7 +18,7 @@ from sqlalchemy import ( func, ForeignKey, DateTime, - PassiveDefault, + DefaultClause, Boolean, ) from sqlalchemy.dialects.postgresql import JSON @@ -39,7 +39,7 @@ class Authority(db.Model): plugin_name = Column(String(64)) description = Column(Text) options = Column(JSON) - date_created = Column(DateTime, PassiveDefault(func.now()), nullable=False) + date_created = Column(DateTime, DefaultClause(func.now()), nullable=False) roles = relationship( "Role", secondary=roles_authorities, diff --git a/lemur/certificates/models.py b/lemur/certificates/models.py index f6562b3f..94e3a42e 100644 --- a/lemur/certificates/models.py +++ b/lemur/certificates/models.py @@ -16,7 +16,7 @@ from sqlalchemy import ( Integer, ForeignKey, String, - PassiveDefault, + DefaultClause, func, Column, Text, @@ -138,7 +138,7 @@ class Certificate(db.Model): not_after = Column(ArrowType) not_after_ix = Index("ix_certificates_not_after", not_after.desc()) - date_created = Column(ArrowType, PassiveDefault(func.now()), nullable=False) + date_created = Column(ArrowType, DefaultClause(func.now()), nullable=False) signing_algorithm = Column(String(128)) status = Column(String(128)) @@ -184,7 +184,6 @@ class Certificate(db.Model): "PendingCertificate", secondary=pending_cert_replacement_associations, backref="pending_replace", - viewonly=True, ) logs = relationship("Log", backref="certificate") diff --git a/lemur/certificates/service.py b/lemur/certificates/service.py index 167425cc..ac844120 100644 --- a/lemur/certificates/service.py +++ b/lemur/certificates/service.py @@ -12,6 +12,7 @@ from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization from flask import current_app from sqlalchemy import func, or_, not_, cast, Integer +from sqlalchemy.sql.expression import false, true from lemur import database from lemur.authorities.models import Authority @@ -150,7 +151,7 @@ def get_all_certs_attached_to_endpoint_without_autorotate(): """ return ( Certificate.query.filter(Certificate.endpoints.any()) - .filter(Certificate.rotation == False) + .filter(Certificate.rotation == false()) .filter(Certificate.not_after >= arrow.now()) .filter(not_(Certificate.replaced.any())) .all() # noqa @@ -205,9 +206,9 @@ def get_all_pending_reissue(): :return: """ return ( - Certificate.query.filter(Certificate.rotation == True) + Certificate.query.filter(Certificate.rotation == true()) .filter(not_(Certificate.replaced.any())) - .filter(Certificate.in_rotation_window == True) + .filter(Certificate.in_rotation_window == true()) .all() ) # noqa @@ -525,7 +526,7 @@ def render(args): ) if current_app.config.get("ALLOW_CERT_DELETION", False): - query = query.filter(Certificate.deleted == False) # noqa + query = query.filter(Certificate.deleted == false()) result = database.sort_and_page(query, Certificate, args) return result diff --git a/lemur/certificates/utils.py b/lemur/certificates/utils.py index e642e058..18f1c967 100644 --- a/lemur/certificates/utils.py +++ b/lemur/certificates/utils.py @@ -82,4 +82,4 @@ def get_key_type_from_csr(data): raise Exception("Unsupported key type") except NotImplemented: - raise NotImplemented() + raise NotImplementedError diff --git a/lemur/factory.py b/lemur/factory.py index 0563d873..edea571a 100644 --- a/lemur/factory.py +++ b/lemur/factory.py @@ -10,7 +10,7 @@ """ import os -import imp +import importlib import errno import pkg_resources import socket @@ -73,8 +73,9 @@ def from_file(file_path, silent=False): :param file_path: :param silent: """ - d = imp.new_module("config") - d.__file__ = file_path + module_spec = importlib.util.spec_from_file_location("config", file_path) + d = importlib.util.module_from_spec(module_spec) + try: with open(file_path) as config_file: exec( # nosec: config file safe diff --git a/lemur/logs/models.py b/lemur/logs/models.py index 07a2ded3..30cc204a 100644 --- a/lemur/logs/models.py +++ b/lemur/logs/models.py @@ -7,7 +7,7 @@ .. moduleauthor:: Kevin Glisson """ -from sqlalchemy import Column, Integer, ForeignKey, PassiveDefault, func, Enum +from sqlalchemy import Column, Integer, ForeignKey, DefaultClause, func, Enum from sqlalchemy_utils.types.arrow import ArrowType @@ -29,5 +29,5 @@ class Log(db.Model): ), nullable=False, ) - logged_at = Column(ArrowType(), PassiveDefault(func.now()), nullable=False) + logged_at = Column(ArrowType(), DefaultClause(func.now()), nullable=False) user_id = Column(Integer, ForeignKey("users.id"), nullable=False) diff --git a/lemur/notifications/messaging.py b/lemur/notifications/messaging.py index 3928689e..75d227b1 100644 --- a/lemur/notifications/messaging.py +++ b/lemur/notifications/messaging.py @@ -16,6 +16,7 @@ from itertools import groupby import arrow from flask import current_app from sqlalchemy import and_ +from sqlalchemy.sql.expression import false, true from lemur import database from lemur.certificates.models import Certificate @@ -40,10 +41,10 @@ def get_certificates(exclude=None): q = ( database.db.session.query(Certificate) .filter(Certificate.not_after <= max) - .filter(Certificate.notify == True) - .filter(Certificate.expired == False) - .filter(Certificate.revoked == False) - ) # noqa + .filter(Certificate.notify == true()) + .filter(Certificate.expired == false()) + .filter(Certificate.revoked == false()) + ) exclude_conditions = [] if exclude: @@ -137,11 +138,11 @@ def send_expiration_notifications(exclude): # security team gets all security_email = current_app.config.get("LEMUR_SECURITY_TEAM_EMAIL") - security_data = [] for owner, notification_group in get_eligible_certificates(exclude=exclude).items(): for notification_label, certificates in notification_group.items(): notification_data = [] + security_data = [] notification = certificates[0][0] diff --git a/lemur/notifications/service.py b/lemur/notifications/service.py index 34edccc0..5bc5f3e1 100644 --- a/lemur/notifications/service.py +++ b/lemur/notifications/service.py @@ -43,7 +43,7 @@ def create_default_expiration_notifications(name, recipients, intervals=None): "name": "recipients", "type": "str", "required": True, - "validation": "^([\w+-.%]+@[\w-.]+\.[A-Za-z]{2,4},?)+$", + "validation": r"^([\w+-.%]+@[\w-.]+\.[A-Za-z]{2,4},?)+$", "helpMessage": "Comma delimited list of email addresses", "value": ",".join(recipients), }, @@ -63,7 +63,7 @@ def create_default_expiration_notifications(name, recipients, intervals=None): "name": "interval", "type": "int", "required": True, - "validation": "^\d+$", + "validation": r"^\d+$", "helpMessage": "Number of days to be alert before expiration.", "value": i, } diff --git a/lemur/pending_certificates/models.py b/lemur/pending_certificates/models.py index fa6be073..ee3e5e97 100644 --- a/lemur/pending_certificates/models.py +++ b/lemur/pending_certificates/models.py @@ -9,7 +9,7 @@ from sqlalchemy import ( Integer, ForeignKey, String, - PassiveDefault, + DefaultClause, func, Column, Text, @@ -76,14 +76,14 @@ class PendingCertificate(db.Model): chain = Column(Text()) private_key = Column(Vault, nullable=True) - date_created = Column(ArrowType, PassiveDefault(func.now()), nullable=False) + date_created = Column(ArrowType, DefaultClause(func.now()), nullable=False) dns_provider_id = Column( Integer, ForeignKey("dns_providers.id", ondelete="CASCADE") ) status = Column(Text(), nullable=True) last_updated = Column( - ArrowType, PassiveDefault(func.now()), onupdate=func.now(), nullable=False + ArrowType, DefaultClause(func.now()), onupdate=func.now(), nullable=False ) rotation = Column(Boolean, default=False) diff --git a/lemur/plugins/bases/notification.py b/lemur/plugins/bases/notification.py index 0da0dad2..76aa33de 100644 --- a/lemur/plugins/bases/notification.py +++ b/lemur/plugins/bases/notification.py @@ -42,7 +42,7 @@ class ExpirationNotificationPlugin(NotificationPlugin): "name": "interval", "type": "int", "required": True, - "validation": "^\d+$", + "validation": r"^\d+$", "helpMessage": "Number of days to be alert before expiration.", }, { diff --git a/lemur/plugins/lemur_acme/plugin.py b/lemur/plugins/lemur_acme/plugin.py index 792e7910..51f6aa3a 100644 --- a/lemur/plugins/lemur_acme/plugin.py +++ b/lemur/plugins/lemur_acme/plugin.py @@ -46,7 +46,7 @@ class ACMEIssuerPlugin(IssuerPlugin): "name": "acme_url", "type": "str", "required": True, - "validation": "/^http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+$/", + "validation": r"/^http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+$/", "helpMessage": "Must be a valid web url starting with http[s]://", }, { @@ -59,7 +59,7 @@ class ACMEIssuerPlugin(IssuerPlugin): "name": "email", "type": "str", "default": "", - "validation": "/^?([-a-zA-Z0-9.`?{}]+@\w+\.\w+)$/", + "validation": r"/^?([-a-zA-Z0-9.`?{}]+@\w+\.\w+)$/", "helpMessage": "Email to use", }, { diff --git a/lemur/plugins/lemur_acme/tests/test_acme_dns.py b/lemur/plugins/lemur_acme/tests/test_acme_dns.py index 3efc3934..acedf977 100644 --- a/lemur/plugins/lemur_acme/tests/test_acme_dns.py +++ b/lemur/plugins/lemur_acme/tests/test_acme_dns.py @@ -3,6 +3,7 @@ from unittest.mock import patch, Mock import josepy as jose from cryptography.x509 import DNSName +from flask import Flask from lemur.plugins.lemur_acme import plugin from lemur.plugins.lemur_acme.acme_handlers import AuthorizationRecord from lemur.common.utils import generate_private_key @@ -23,6 +24,16 @@ class TestAcmeDns(unittest.TestCase): "test.fakedomain.net": [mock_dns_provider], } + # Creates a new Flask application for a test duration. In python 3.8, manual push of application context is + # needed to run tests in dev environment without getting error 'Working outside of application context'. + _app = Flask('lemur_test_acme') + self.ctx = _app.app_context() + assert self.ctx + self.ctx.push() + + def tearDown(self): + self.ctx.pop() + @patch("lemur.plugins.lemur_acme.plugin.len", return_value=1) def test_get_dns_challenges(self, mock_len): assert mock_len @@ -105,22 +116,24 @@ class TestAcmeDns(unittest.TestCase): mock_dns_provider = Mock() mock_dns_provider.wait_for_dns_change = Mock(return_value=True) + mock_dns_challenge = Mock() + response = Mock() + response.simple_verify = Mock(return_value=False) + mock_dns_challenge.response = Mock(return_value=response) + mock_authz = Mock() - mock_authz.dns_challenge.response = Mock() - mock_authz.dns_challenge.response.simple_verify = Mock(return_value=False) - mock_authz.authz = [] + mock_authz.dns_challenge = [] + mock_authz.dns_challenge.append(mock_dns_challenge) + mock_authz.target_domain = "www.test.com" mock_authz_record = Mock() mock_authz_record.body.identifier.value = "test" + mock_authz.authz = [] mock_authz.authz.append(mock_authz_record) mock_authz.change_id = [] mock_authz.change_id.append("123") - mock_authz.dns_challenge = [] - dns_challenge = Mock() - mock_authz.dns_challenge.append(dns_challenge) - self.assertRaises( - ValueError, self.acme.complete_dns_challenge(mock_acme, mock_authz) - ) + with self.assertRaises(ValueError): + self.acme.complete_dns_challenge(mock_acme, mock_authz) @patch("acme.client.Client") @patch("OpenSSL.crypto", return_value="mock_cert") diff --git a/lemur/plugins/lemur_acme/tests/test_powerdns.py b/lemur/plugins/lemur_acme/tests/test_powerdns.py index 37e4968e..cf850970 100644 --- a/lemur/plugins/lemur_acme/tests/test_powerdns.py +++ b/lemur/plugins/lemur_acme/tests/test_powerdns.py @@ -1,5 +1,7 @@ import unittest from unittest.mock import patch, Mock + +from flask import Flask from lemur.plugins.lemur_acme import plugin, powerdns @@ -17,6 +19,16 @@ class TestPowerdns(unittest.TestCase): "test.fakedomain.net": [mock_dns_provider], } + # Creates a new Flask application for a test duration. In python 3.8, manual push of application context is + # needed to run tests in dev environment without getting error 'Working outside of application context'. + _app = Flask('lemur_test_acme') + self.ctx = _app.app_context() + assert self.ctx + self.ctx.push() + + def tearDown(self): + self.ctx.pop() + @patch("lemur.plugins.lemur_acme.powerdns.current_app") def test_get_zones(self, mock_current_app): account_number = "1234567890" diff --git a/lemur/plugins/lemur_acme/tests/test_ultradns.py b/lemur/plugins/lemur_acme/tests/test_ultradns.py index f1d61e68..7616459e 100644 --- a/lemur/plugins/lemur_acme/tests/test_ultradns.py +++ b/lemur/plugins/lemur_acme/tests/test_ultradns.py @@ -1,6 +1,7 @@ import unittest from unittest.mock import patch, Mock +from flask import Flask from lemur.plugins.lemur_acme import plugin, ultradns from requests.models import Response @@ -19,6 +20,16 @@ class TestUltradns(unittest.TestCase): "test.fakedomain.net": [mock_dns_provider], } + # Creates a new Flask application for a test duration. In python 3.8, manual push of application context is + # needed to run tests in dev environment without getting error 'Working outside of application context'. + _app = Flask('lemur_test_acme') + self.ctx = _app.app_context() + assert self.ctx + self.ctx.push() + + def tearDown(self): + self.ctx.pop() + @patch("lemur.plugins.lemur_acme.ultradns.requests") @patch("lemur.plugins.lemur_acme.ultradns.current_app") def test_ultradns_get_token(self, mock_current_app, mock_requests): diff --git a/lemur/plugins/lemur_aws/plugin.py b/lemur/plugins/lemur_aws/plugin.py index 1be641b0..b54787ac 100644 --- a/lemur/plugins/lemur_aws/plugin.py +++ b/lemur/plugins/lemur_aws/plugin.py @@ -33,6 +33,7 @@ .. moduleauthor:: Harm Weites """ +import sys from acme.errors import ClientError from flask import current_app @@ -408,6 +409,47 @@ class S3DestinationPlugin(ExportDestinationPlugin): account_number=self.get_option("accountNumber", options), ) + def upload_acme_token(self, token_path, token, options, **kwargs): + """ + This is called from the acme http challenge + :param self: + :param token_path: + :param token: + :param options: + :param kwargs: + :return: + """ + current_app.logger.debug("S3 destination plugin is started for HTTP-01 challenge") + + function = f"{__name__}.{sys._getframe().f_code.co_name}" + + account_number = self.get_option("accountNumber", options) + bucket_name = self.get_option("bucket", options) + prefix = self.get_option("prefix", options) + region = self.get_option("region", options) + filename = token_path.split("/")[-1] + if not prefix.endswith("/"): + prefix + "/" + + res = s3.put(bucket_name=bucket_name, + region_name=region, + prefix=prefix + filename, + data=token, + encrypt=False, + account_number=account_number) + res = "Success" if res else "Failure" + log_data = { + "function": function, + "message": "check if any valid certificate is revoked", + "result": res, + "bucket_name": bucket_name, + "filename": filename + } + current_app.logger.info(log_data) + metrics.send(f"{function}", "counter", 1, metric_tags={"result": res, + "bucket_name": bucket_name, + "filename": filename}) + class SNSNotificationPlugin(ExpirationNotificationPlugin): title = "AWS SNS" diff --git a/lemur/plugins/lemur_aws/s3.py b/lemur/plugins/lemur_aws/s3.py index 43faa28f..1b0831b3 100644 --- a/lemur/plugins/lemur_aws/s3.py +++ b/lemur/plugins/lemur_aws/s3.py @@ -6,12 +6,15 @@ :license: Apache, see LICENSE for more details. .. moduleauthor:: Kevin Glisson """ +from botocore.exceptions import ClientError from flask import current_app +from lemur.extensions import sentry + from .sts import sts_client @sts_client("s3", service_type="resource") -def put(bucket_name, region, prefix, data, encrypt, **kwargs): +def put(bucket_name, region_name, prefix, data, encrypt, **kwargs): """ Use STS to write to an S3 bucket """ @@ -32,4 +35,41 @@ def put(bucket_name, region, prefix, data, encrypt, **kwargs): ServerSideEncryption="AES256", ) else: - bucket.put_object(Key=prefix, Body=data, ACL="bucket-owner-full-control") + try: + bucket.put_object(Key=prefix, Body=data, ACL="bucket-owner-full-control") + return True + except ClientError: + sentry.captureException() + return False + + +@sts_client("s3", service_type="client") +def delete(bucket_name, prefixed_object_name, **kwargs): + """ + Use STS to delete an object + """ + try: + response = kwargs["client"].delete_object(Bucket=bucket_name, Key=prefixed_object_name) + current_app.logger.debug(f"Delete data from S3." + f"Bucket: {bucket_name}," + f"Prefix: {prefixed_object_name}," + f"Status_code: {response}") + return response['ResponseMetadata']['HTTPStatusCode'] < 300 + except ClientError: + sentry.captureException() + return False + + +@sts_client("s3", service_type="client") +def get(bucket_name, prefixed_object_name, **kwargs): + """ + Use STS to get an object + """ + try: + response = kwargs["client"].get_object(Bucket=bucket_name, Key=prefixed_object_name) + current_app.logger.debug(f"Get data from S3. Bucket: {bucket_name}," + f"object_name: {prefixed_object_name}") + return response['Body'].read().decode("utf-8") + except ClientError: + sentry.captureException() + return None diff --git a/lemur/plugins/lemur_aws/tests/test_plugin.py b/lemur/plugins/lemur_aws/tests/test_plugin.py index dbad7b02..be9b14fd 100644 --- a/lemur/plugins/lemur_aws/tests/test_plugin.py +++ b/lemur/plugins/lemur_aws/tests/test_plugin.py @@ -1,5 +1,82 @@ +import boto3 +from moto import mock_sts, mock_s3 + + def test_get_certificates(app): from lemur.plugins.base import plugins p = plugins.get("aws-s3") assert p + + +@mock_sts() +@mock_s3() +def test_upload_acme_token(app): + from lemur.plugins.base import plugins + from lemur.plugins.lemur_aws.s3 import get + + bucket = "public-bucket" + account = "123456789012" + prefix = "some-path/more-path/" + token_content = "Challenge" + token_name = "TOKEN" + token_path = ".well-known/acme-challenge/" + token_name + + additional_options = [ + { + "name": "bucket", + "value": bucket, + "type": "str", + "required": True, + "validation": r"[0-9a-z.-]{3,63}", + "helpMessage": "Must be a valid S3 bucket name!", + }, + { + "name": "accountNumber", + "type": "str", + "value": account, + "required": True, + "validation": r"[0-9]{12}", + "helpMessage": "A valid AWS account number with permission to access S3", + }, + { + "name": "region", + "type": "str", + "default": "us-east-1", + "required": False, + "helpMessage": "Region bucket exists", + "available": ["us-east-1", "us-west-2", "eu-west-1"], + }, + { + "name": "encrypt", + "type": "bool", + "value": False, + "required": False, + "helpMessage": "Enable server side encryption", + "default": True, + }, + { + "name": "prefix", + "type": "str", + "value": prefix, + "required": False, + "helpMessage": "Must be a valid S3 object prefix!", + }, + ] + + s3_client = boto3.client('s3') + s3_client.create_bucket(Bucket=bucket) + p = plugins.get("aws-s3") + + p.upload_acme_token(token_path=token_path, + token_content=token_content, + token=token_content, + options=additional_options) + + response = get(bucket_name=bucket, + prefixed_object_name=prefix + token_name, + encrypt=False, + account_number=account) + + # put data, and getting the same data + assert (response == token_content) diff --git a/lemur/plugins/lemur_aws/tests/test_s3.py b/lemur/plugins/lemur_aws/tests/test_s3.py new file mode 100644 index 00000000..7d0fa843 --- /dev/null +++ b/lemur/plugins/lemur_aws/tests/test_s3.py @@ -0,0 +1,41 @@ +import boto3 +from moto import mock_sts, mock_s3 + + +@mock_sts() +@mock_s3() +def test_put_delete_s3_object(app): + from lemur.plugins.lemur_aws.s3 import put, delete, get + + bucket = "public-bucket" + region = "us-east-1" + account = "123456789012" + path = "some-path/foo" + data = "dummy data" + + s3_client = boto3.client('s3') + s3_client.create_bucket(Bucket=bucket) + + put(bucket_name=bucket, + region_name=region, + prefix=path, + data=data, + encrypt=False, + account_number=account, + region=region) + + response = get(bucket_name=bucket, prefixed_object_name=path, account_number=account) + + # put data, and getting the same data + assert (response == data) + + response = get(bucket_name="wrong-bucket", prefixed_object_name=path, account_number=account) + + # attempting to get thccle wrong data + assert (response is None) + + delete(bucket_name=bucket, prefixed_object_name=path, account_number=account) + response = get(bucket_name=bucket, prefixed_object_name=path, account_number=account) + + # delete data, and getting the same data + assert (response is None) diff --git a/lemur/plugins/lemur_email/plugin.py b/lemur/plugins/lemur_email/plugin.py index f380c82e..041b27ec 100644 --- a/lemur/plugins/lemur_email/plugin.py +++ b/lemur/plugins/lemur_email/plugin.py @@ -91,7 +91,7 @@ class EmailNotificationPlugin(ExpirationNotificationPlugin): "name": "recipients", "type": "str", "required": True, - "validation": "^([\w+-.%]+@[\w-.]+\.[A-Za-z]{2,4},?)+$", + "validation": r"^([\w+-.%]+@[\w-.]+\.[A-Za-z]{2,4},?)+$", "helpMessage": "Comma delimited list of email addresses", } ] diff --git a/lemur/plugins/lemur_sftp/plugin.py b/lemur/plugins/lemur_sftp/plugin.py index 1c974a28..7f3ccc4f 100644 --- a/lemur/plugins/lemur_sftp/plugin.py +++ b/lemur/plugins/lemur_sftp/plugin.py @@ -48,7 +48,7 @@ class SFTPDestinationPlugin(DestinationPlugin): "type": "int", "required": True, "helpMessage": "The SFTP port, default is 22.", - "validation": "^(6553[0-5]|655[0-2][0-9]\d|65[0-4](\d){2}|6[0-4](\d){3}|[1-5](\d){4}|[1-9](\d){0,3})", + "validation": r"^(6553[0-5]|655[0-2][0-9]\d|65[0-4](\d){2}|6[0-4](\d){3}|[1-5](\d){4}|[1-9](\d){0,3})", "default": "22", }, { diff --git a/lemur/plugins/lemur_slack/plugin.py b/lemur/plugins/lemur_slack/plugin.py index 70d97aa5..3ad22bca 100644 --- a/lemur/plugins/lemur_slack/plugin.py +++ b/lemur/plugins/lemur_slack/plugin.py @@ -89,7 +89,7 @@ class SlackNotificationPlugin(ExpirationNotificationPlugin): "name": "webhook", "type": "str", "required": True, - "validation": "^https:\/\/hooks\.slack\.com\/services\/.+$", + "validation": r"^https:\/\/hooks\.slack\.com\/services\/.+$", "helpMessage": "The url Slack told you to use for this integration", }, { diff --git a/lemur/tests/test_dns_providers.py b/lemur/tests/test_dns_providers.py index 83315be5..9b8fdb5a 100644 --- a/lemur/tests/test_dns_providers.py +++ b/lemur/tests/test_dns_providers.py @@ -13,7 +13,7 @@ class TestDNSProvider(unittest.TestCase): self.assertFalse(dnsutil.is_valid_domain('example-of-over-63-character-domain-label-length-limit-123456789.com')) self.assertTrue(dnsutil.is_valid_domain('_acme-chall.example.com')) self.assertFalse(dnsutil.is_valid_domain('e/xample.com')) - self.assertFalse(dnsutil.is_valid_domain('exam\ple.com')) + self.assertFalse(dnsutil.is_valid_domain('exam\\ple.com')) self.assertFalse(dnsutil.is_valid_domain('