Branch was auto-updated.

This commit is contained in:
pyth0n1c
2022-03-22 13:21:54 -07:00
committed by GitHub
105 changed files with 41 additions and 526946 deletions
+1 -1
View File
@@ -583,4 +583,4 @@ jobs:
if [ "$API_STATUS" != "200" ]; then
echo "Error [Security Content API status: $API_STATUS]"
exit 1
fi
fi
+40 -40
View File
@@ -62,7 +62,7 @@ jobs:
- name: Install Python Dependencies
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install wheel
@@ -70,7 +70,7 @@ jobs:
- name: Run the CI
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
source .venv/bin/activate
echo "github.event.issue.pull_request : [${{ github.event.issue.pull_request }}]"
echo "github.event.pull_request.number : [${{ github.event.pull_request.number }}]"
@@ -97,24 +97,24 @@ jobs:
with:
name: testing-results-config
path: |
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/apps/DA-ESS-ContentUpdate-latest.tar.gz
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_0.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_1.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_2.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_3.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_4.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_5.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_6.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_7.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_8.json
bin/automated_detection_testing/ci/detection_testing_batch/prior_config/config_tests_9.json
bin/docker_detection_tester/prior_config/apps/DA-ESS-ContentUpdate-latest.tar.gz
bin/docker_detection_tester/prior_config/config_tests_0.json
bin/docker_detection_tester/prior_config/config_tests_1.json
bin/docker_detection_tester/prior_config/config_tests_2.json
bin/docker_detection_tester/prior_config/config_tests_3.json
bin/docker_detection_tester/prior_config/config_tests_4.json
bin/docker_detection_tester/prior_config/config_tests_5.json
bin/docker_detection_tester/prior_config/config_tests_6.json
bin/docker_detection_tester/prior_config/config_tests_7.json
bin/docker_detection_tester/prior_config/config_tests_8.json
bin/docker_detection_tester/prior_config/config_tests_9.json
- name: Upload File to Enable Replication of the Test at a Different Time or Place
uses: actions/upload-artifact@v2
with:
name: replicate_test
path: |
bin/automated_detection_testing/ci/detection_testing_batch/replicate_test.json
bin/docker_detection_tester/replicate_test.json
docker-detection-testing-execution:
runs-on: ubuntu-latest
@@ -146,7 +146,7 @@ jobs:
uses: actions/download-artifact@v2
with:
name: testing-results-config
path: bin/automated_detection_testing/ci/detection_testing_batch/prior_config
path: bin/docker_detection_tester/prior_config
- uses: actions/setup-python@v2
@@ -156,7 +156,7 @@ jobs:
- name: Install Python Dependencies
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install wheel
@@ -164,7 +164,7 @@ jobs:
- name: Run the CI
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
source .venv/bin/activate
python3 detection_testing_execution.py run -c prior_config/${{ matrix.manifest_filename}}
@@ -175,16 +175,16 @@ jobs:
with:
name: ${{ matrix.manifest_filename}}.results
path: |
bin/automated_detection_testing/ci/detection_testing_batch/test_results/success.csv
bin/automated_detection_testing/ci/detection_testing_batch/test_results/error.csv
bin/automated_detection_testing/ci/detection_testing_batch/test_results/failure.csv
bin/automated_detection_testing/ci/detection_testing_batch/test_results/combined.csv
bin/automated_detection_testing/ci/detection_testing_batch/test_results/success.json
bin/automated_detection_testing/ci/detection_testing_batch/test_results/error.json
bin/automated_detection_testing/ci/detection_testing_batch/test_results/failure.json
bin/automated_detection_testing/ci/detection_testing_batch/test_results/combined.json
bin/docker_detection_tester/test_results/success.csv
bin/docker_detection_tester/test_results/error.csv
bin/docker_detection_tester/test_results/failure.csv
bin/docker_detection_tester/test_results/combined.csv
bin/docker_detection_tester/test_results/success.json
bin/docker_detection_tester/test_results/error.json
bin/docker_detection_tester/test_results/failure.json
bin/docker_detection_tester/test_results/combined.json
bin/automated_detection_testing/ci/detection_testing_batch/test_results/summary.json
bin/docker_detection_tester/test_results/summary.json
docker-detection-testing-execution-merge-results:
runs-on: ubuntu-latest
@@ -205,52 +205,52 @@ jobs:
uses: actions/download-artifact@v2
with:
name: config_tests_0.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_0
path: bin/docker_detection_tester/results_0
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_1.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_1
path: bin/docker_detection_tester/results_1
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_2.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_2
path: bin/docker_detection_tester/results_2
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_3.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_3
path: bin/docker_detection_tester/results_3
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_4.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_4
path: bin/docker_detection_tester/results_4
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_5.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_5
path: bin/docker_detection_tester/results_5
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_6.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_6
path: bin/docker_detection_tester/results_6
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_7.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_7
path: bin/docker_detection_tester/results_7
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_8.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_8
path: bin/docker_detection_tester/results_8
- name: Download artifacts
uses: actions/download-artifact@v2
with:
name: config_tests_9.json.results
path: bin/automated_detection_testing/ci/detection_testing_batch/results_9
path: bin/docker_detection_tester/results_9
- uses: actions/setup-python@v2
with:
@@ -259,7 +259,7 @@ jobs:
- name: Install Python Dependencies
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install wheel
@@ -267,7 +267,7 @@ jobs:
- name: Merge Detections into single File
run: |
cd bin/automated_detection_testing/ci/detection_testing_batch
cd bin/docker_detection_tester
source .venv/bin/activate
python summarize_json.py --files results_*/summary.json --output_filename summary_test_results.json
@@ -278,7 +278,7 @@ jobs:
with:
name: SummaryTestResults
path: |
bin/automated_detection_testing/ci/detection_testing_batch/summary_test_results.json
bin/docker_detection_tester
- name: Upload Failures Manifest on Failure
uses: actions/upload-artifact@v2
@@ -286,7 +286,7 @@ jobs:
with:
name: DetectionFailureManifest
path: |
bin/automated_detection_testing/ci/detection_testing_batch/detection_failure_manifest.json
bin/docker_detection_tester/detection_failure_manifest.json
#Always clean these up, they make the output messy
-60
View File
@@ -1,66 +1,6 @@
variables:
GIT_SUBMODULE_STRATEGY: recursive
K8S_PLAYGROUND_ROLE: k8s:v1/app-play1/namespace/threat-research/role/deployer
K8S_GSTAGING_ROLE: k8s:v1/kube-wicket-iad10/namespace/tr-st-deploy/role/deployer
DOCKER_ROLE: docker:v1/registry/prefix/threat-research
K8_DEPLOYER_CONTAINER: docker.repo.splunkdev.net/threat-research/k8-deployer
SMOKETEST_RUNNER: docker.repo.splunkdev.net/threat-research/smoketest-runner
SRCBRANCH: $CI_COMMIT_REF_NAME
stages:
- publish_smoketest_runner
- publish_deployer
- smoketest_staging
- publish_build_to_pre_qa
publish_deployer:
stage: publish_deployer
image: docker.repo.splunkdev.net/ci-cd/ci-container:alpine-3.11
before_script:
- apk add --update docker
script:
- eval $(go-go vault -a ${DOCKER_ROLE})
- cd bin/ssa-end-to-end-testing/k8s-deployer
- docker build . -t ${K8_DEPLOYER_CONTAINER}:${CI_COMMIT_SHORT_SHA}
- docker push ${K8_DEPLOYER_CONTAINER}:${CI_COMMIT_SHORT_SHA}
only:
- /^ssa.*$/
- develop
publish_smoketest_runner:
stage: publish_smoketest_runner
image: docker.repo.splunkdev.net/ci-cd/ci-container:alpine-3.11
before_script:
- apk add --update docker
script:
- eval $(go-go vault -a ${DOCKER_ROLE})
- docker build bin/ssa-end-to-end-testing/smoke-test-runner -t ${SMOKETEST_RUNNER}:${CI_COMMIT_SHORT_SHA} --build-arg SRCBRANCH=$CI_COMMIT_REF_NAME
- docker push ${SMOKETEST_RUNNER}:${CI_COMMIT_SHORT_SHA}
only:
- /^ssa.*$/
- develop
smoketest_staging:
stage: smoketest_staging
resource_group: staging
dependencies:
- publish_smoketest_runner
- publish_deployer
image: ${K8_DEPLOYER_CONTAINER}:${CI_COMMIT_SHORT_SHA}
script:
- eval $(go-go vault -a ${K8S_GSTAGING_ROLE})
- eval $(go-go vault -a ${DOCKER_ROLE})
- cd /deployer/k8s
- ./deployer.sh
- ./check_deploy_status.sh
variables:
SCSENV: app_gstaging1
SMOKETEST_RUNNER_IMAGE: ${SMOKETEST_RUNNER}:${CI_COMMIT_SHORT_SHA}
only:
- /^ssa.*$/
- develop
publish_build_to_pre_qa:
stage: publish_build_to_pre_qa
artifacts:
@@ -1,6 +0,0 @@
venv
security-content
attack_range
attack_range.log
Dockerfile
@@ -1,22 +0,0 @@
FROM ubuntu:18.04
MAINTAINER Patrick Bareiss
RUN apt-get update
RUN DEBIAN_FRONTEND="noninteractive" apt-get -y install tzdata
RUN apt-get install -y python3-dev git python-dev unzip python3-pip awscli
RUN apt-get install -y python-gitdb
RUN apt-get install -y wget unzip
RUN wget --quiet https://releases.hashicorp.com/terraform/0.13.1/terraform_0.13.1_linux_amd64.zip \
&& unzip terraform_0.13.1_linux_amd64.zip \
&& mv terraform /usr/bin \
&& rm terraform_0.13.1_linux_amd64.zip
ADD config /root/.aws/config
ADD . /app
WORKDIR /app
RUN pip3 install -r requirements.txt
ENTRYPOINT ["python3", "detection_service.py"]
CMD ["-tfn", "T1003_002"]
-152
View File
@@ -1,152 +0,0 @@
# Automated Detection Testing
Testing detection is a very important task during detection engineering. As it takes a lot of time and is a repeating task, we used the Attack Range to build an automated detection testing service.
## Architecture
![Architecture](automated_detection_testing/architecture_automated_detection_testing.png)
The automated detection testing is using AWS Batch as execution engine. AWS batch allows you to run batch computing jobs, in our case a automated detection test. detection_service.py is the executable which controls the detection test. This executable is deployed in a docker container which is used by AWS Batch.
## Usage
```
python detection_service.py
usage: detection_service.py [-h] -tfn TEST_FILE_NAME
[-arr ATTACK_RANGE_REPO]
[-arb ATTACK_RANGE_BRANCH]
[-scr SECURITY_CONTENT_REPO]
[-scb SECURITY_CONTENT_BRANCH]
[-gt GITHUB_TOKEN]
[-smk SECRETS_MANAGER_KEY]
[-s3b S3_BUCKET]
detection_service.py: error: the following arguments are required: -tfn/--test_file_name
```
The detection_service.py has one mandatory parameter, which is --test_file_name. This parameter will look into the security content repository under tests for the specified test file (without extension). The other parameters are optional and can be used to specify forks of projects or specific branches. The detection_service.py is creating Pull Requests after a successful test. Therefore, it needs a Github OAUTH Token. This can be either added with the parameter --github_token or can be derived from the [AWS secrets manager](https://aws.amazon.com/secrets-manager/) through --secrets_manager_key. Let's have a look how to use the detection service after you deployed it.
Let's have a look how to use the attack data service after you deployed it:
### Using AWS CLI
Example 1:
```
aws batch submit-job --job-name detection_test_T1003_001 --job-definition detection_service_job --job-queue detection_service_queue --container-overrides '{"command": ["-tfn", "T1003_001"]}'
```
Example 2:
```
aws batch submit-job --job-name detection_test_T1003_001 --job-definition detection_service_job --job-queue detection_service_queue --container-overrides '{"command": ["-tfn", "T1003_001", "-scr", "P4T12ICK/security-content", "-scb", "develop_detection_T1003", "-smk", "github_token"], "-s3b", "my_detection_test_bucket"}'
```
### Using AWS Web Portal
The Attack Data Generation Service can be also triggered over the AWS Web Portal. You will first click on the service "Batch" and then click on the left side "Jobs". Then, you click on "submit new job". You will fill the variables according to the following screenshot and click on "Submit".
![AWS Batch Job](attack_data_service/static/aws_batch_submit_job.png)
## Deployment
In order to deploy the Detection Testing Service to AWS Batch, please follow this guideline. This description assumes that you will deploy the Detection Testing Service to the region eu-central-1.
### Prerequisites
- AWS account
- IAM user with administrative permissions
- AWS CLI
- Docker
- S3 bucket to store detection test data
### Create GitHub Token
The GitHub Token allows the Automate Detection Testing Service to create Pull Requests.
- Create a Personal GitHub Acces Token according to the following [tutorial](https://docs.github.com/en/free-pro-team@latest/github/authenticating-to-github/creating-a-personal-access-token)
### Upload GitHub Token to AWS Secrets Manager
- Connect to AWS Web Portal
- Go to the AWS Secrets Manager
- Choose region eu-central-1
- Click on "Store a new secret"
- Click on "Other type of secrets"
- Add "github_token" as key
- Copy the github token as value
- Click on "Next"
- Use "github_token" as Secret name
- Click on "Next"
- Click on "Next"
- Click on "Store"
### Create AWS ECR Repository
- Connect to AWS Web Portal
- Go to service "Elastic Container Registry"
- Click on "Repositories" under Amazon ECR on the left side.
- Click on "Create repository"
- Add "awsbatch/detection-testing-service" as repository name
- Click on "Create repository"
### Build and Upload Docker File
- Navgigate to the automated_detection_testing folder:
```
cd automated_detection_testing
```
- Build the docker container
```
docker build --tag awsbatch/detection-testing-service .
```
- Tag the docker container (The aws account number can be found in the AWS ECR Repository path)
```
docker tag awsbatch/detection-testing-service:latest [aws_account_number].dkr.ecr.eu-central-1.amazonaws.com/awsbatch/detection-testing-service:latest
```
- Login to AWS ECR
```
aws ecr get-login-password --region eu-central-1 | docker login --username AWS --password-stdin [aws_account_number].dkr.ecr.eu-central-1.amazonaws.com
```
- Upload Docker container
```
docker push [aws_account_number].dkr.ecr.eu-central-1.amazonaws.com/awsbatch/detection-testing-service:latest
```
### Configure AWS Batch
- Connect to AWS Web Portal
- Go to service "AWS Batch"
- Click on "Compute environments" on the left side
- Click on "Create"
- Use "detection_testing_service_environment" as "Compute environment name"
- Define Instance Configuration according to your demand. You can choose small instance types, because the instance will run docker and docker will only run a python script.
- Define the vpc and subnets which you want to use in Networking
- Click on "create compute environment"
- Click on "Job queues" on the left side
- Click on "Create"
- Use "detection_testing_service_queue" as "Job queue name"
- Select "detection_testing_service_environment" as "compute environment"
- Click on "Create"
- Go to service "IAM"
- Create the following role with name: detection_testing_service_role with the Policies AmazonEC2FullAccess, SecretsManagerReadWrite and AmazonS3FullAccess
- Go to service "AWS Batch"
- Click on "Job definitions" on the left side
- Click on "Create"
- Use "detection_testing_service" as Name
- Use 3000 as "Execution timeout"
- Container properties:
- Use "[aws_account_number].dkr.ecr.eu-central-1.amazonaws.com/awsbatch/detection-testing-service:latest" as Image
- remove Command from Command field
- Use 2 in vCPUs
- Use 2048 in Memory
- Click on "Additional configuration"
- Use "detection_testing_service_role" as Job Role
- Use root as "User" under Security
- Click on "Create"
## Local Detection Testing
The Detection Testing Service can be also run locally.
- Navgigate to the automated_detection_testing folder:
```
cd automated_detection_testing
```
- Build the docker container
```
docker build --tag awsbatch/detection-testing-service .
```
- Run the docker container
```
docker run -v ~/.aws/credentials:/root/.aws/credentials:ro --name attackrange awsbatch/detection-testing-service:latest -tfn T1003_001 -s3b my_detection_test_bucket -scr P4T12ICK/security-content -scb new_detections
```
## Troubleshooting
AWS Batch will store the logs in Cloudwatch. Check the cloudwatch logs for Troubleshooting.
Binary file not shown.

Before

Width:  |  Height:  |  Size: 78 KiB

@@ -1,20 +0,0 @@
FROM ubuntu:18.04
RUN apt-get update
RUN DEBIAN_FRONTEND="noninteractive" apt-get -y install tzdata
RUN apt-get install -y python3-dev git python-dev unzip python3-pip awscli
RUN apt-get install -y python-gitdb
RUN apt-get install -y wget unzip
RUN wget --quiet https://releases.hashicorp.com/terraform/0.13.1/terraform_0.13.1_linux_amd64.zip \
&& unzip terraform_0.13.1_linux_amd64.zip \
&& mv terraform /usr/bin \
&& rm terraform_0.13.1_linux_amd64.zip
ADD . /app
WORKDIR /app
RUN pip3 install -r requirements.txt
ENTRYPOINT ["python3", "detection_service.py"]
CMD ["-a", "rebuild"]
@@ -1,58 +0,0 @@
import os
from os import path
import sys
import argparse
import time
import shutil
from helpers import github_service, aws_service, attack_range_controller
DT_ATTACK_RANGE_STATE_STORE = "dt-attack-range-tf-state-store"
DT_ATTACK_RANGE_STATE = "dt-attack-range-state"
REGION = "eu-central-1"
NAME = "detection-testing-attack-range"
def main(args):
parser = argparse.ArgumentParser(description="CI Detection Testing")
parser.add_argument("-a", "--action", required=True, help="action")
args = parser.parse_args()
action = args.action
if action == "build":
build_dt_attack_range()
elif action == "destroy":
destroy_dt_attack_range()
elif action == "rebuild":
destroy_dt_attack_range()
shutil.rmtree('attack_range')
time.sleep(60)
build_dt_attack_range()
def build_dt_attack_range():
github_service.clone_attack_range_project()
aws_service.create_tf_state_store(DT_ATTACK_RANGE_STATE_STORE, REGION)
aws_service.create_db_database(DT_ATTACK_RANGE_STATE, REGION)
ssh_key_name, key_material = aws_service.create_key_pair(REGION)
time.sleep(10)
aws_service.create_entry_database(REGION, DT_ATTACK_RANGE_STATE, NAME, "building", ssh_key_name, key_material)
password = attack_range_controller.build_attack_range(REGION, DT_ATTACK_RANGE_STATE_STORE, ssh_key_name)
aws_service.update_entry_database(REGION, DT_ATTACK_RANGE_STATE, NAME, password, "running")
def destroy_dt_attack_range():
data = aws_service.get_entry_database(REGION, DT_ATTACK_RANGE_STATE, NAME)
github_service.clone_attack_range_project()
attack_range_controller.destroy_attack_range(REGION, data, DT_ATTACK_RANGE_STATE_STORE)
aws_service.delete_db_database(DT_ATTACK_RANGE_STATE, REGION)
aws_service.delete_tf_state_store(REGION, DT_ATTACK_RANGE_STATE_STORE)
if __name__ == "__main__":
main(sys.argv[1:])
@@ -1,82 +0,0 @@
from shutil import which
import secrets
import string
import os
import logging
import sys
import time
from helpers import aws_service
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
def create_random_password():
alphabet = string.ascii_letters + string.digits
password = ''.join(secrets.choice(alphabet) for i in range(10))
password = '!1' + password + 'n:'
return password
def configure_attack_range(region, tf_state_store, password, ssh_key_name):
sys.path.append(os.path.join(os.getcwd(),'attack_range'))
with open('attack_range/attack_range.conf.template', 'r') as file :
filedata = file.read()
filedata = filedata.replace('attack_range_password = Pl3ase-k1Ll-me:p', 'attack_range_password = ' + password)
filedata = filedata.replace('tf_backend = local', 'tf_backend = remote')
filedata = filedata.replace('tf_backend_name = threat_research_attack_range', 'tf_backend_name = ' + tf_state_store)
filedata = filedata.replace('region = us-west-2', 'region = ' + region)
filedata = filedata.replace('windows_domain_controller = 1', 'windows_domain_controller = 0')
filedata = filedata.replace('windows_server_join_domain = 1', 'windows_server_join_domain = 0')
filedata = filedata.replace('range_name = default', 'range_name = dt')
filedata = filedata.replace('key_name = attack-range-key-pair', 'key_name = ' + ssh_key_name)
filedata = filedata.replace('private_key_path = ~/.ssh/id_rsa', 'private_key_path = ' + str(os.getcwd() + "/" + ssh_key_name))
with open('attack_range/attack_range.conf', 'w') as file:
file.write(filedata)
def build_attack_range(region, tf_state_store, ssh_key_name):
password = create_random_password()
configure_attack_range(region, tf_state_store, password, ssh_key_name)
module = __import__('attack_range')
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'build']
try:
LOGGER.info(f"Build Attack Range")
results = module.main(module.sys.argv)
except Exception as e:
LOGGER.error('Build Error: ' + str(e))
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'destroy']
module.main(module.sys.argv)
sys.exit(1)
return password
def destroy_attack_range(region, data, tf_state_store):
password = data['password']
ssh_key_name = data['ssh_key_name']
configure_attack_range(region, tf_state_store, password, ssh_key_name)
module = __import__('attack_range')
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'destroy']
try:
LOGGER.info(f"Destroy Attack Range")
results = module.main(module.sys.argv)
except Exception as e:
LOGGER.error('Build Error: ' + str(e))
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'destroy']
module.main(module.sys.argv)
sys.exit(1)
@@ -1,167 +0,0 @@
import boto3
from botocore.config import Config
import time
import os
import json
def create_key_pair(region):
my_config = Config(region_name = region)
epoch_time = str(int(time.time()))
ssh_key_name = 'key-dt-' + epoch_time
ec2 = boto3.client('ec2', config=my_config)
response = ec2.create_key_pair(KeyName=ssh_key_name)
with open(ssh_key_name, "w") as ssh_key:
ssh_key.write(response['KeyMaterial'])
os.chmod(ssh_key_name, 0o600)
private_key_path = str(os.getcwd() + "/" + ssh_key_name)
return ssh_key_name, response['KeyMaterial']
def delete_key_pair(region, key_pair_name):
my_config = Config(region_name = region)
ec2 = boto3.client('ec2', config=my_config)
response = ec2.delete_key_pair(KeyName=key_pair_name)
def create_entry_database(region, db_name, name, state, ssh_key_name, private_key):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table(db_name)
response = table.put_item(Item= {
'name': name,
'ssh_key_name': ssh_key_name,
'private_key': private_key,
'status': state
})
def update_entry_database(region, db_name, name, password, state):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table(db_name)
response = table.update_item(
Key={
'name': name
},
UpdateExpression="set #ts=:s, password=:p",
ExpressionAttributeValues={
':s': state,
':p': password
},
ExpressionAttributeNames={
"#ts": "status"
},
ReturnValues="UPDATED_NEW"
)
def delete_entry_database(region, db_name, name):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table(db_name)
response = table.delete_item(
Key={
'name': name
}
)
def get_entry_database(region, db_name, name):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table(db_name)
response = table.get_item(
Key={
'name': name
}
)
if 'Item' in response:
return response['Item']
else:
return {}
def create_db_database(name, region):
my_config = Config(region_name = region)
client = boto3.client('dynamodb', config=my_config)
response = client.create_table(
TableName=name,
KeySchema=[
{
'AttributeName': 'name',
'KeyType': 'HASH' # Partition key
}
],
AttributeDefinitions=[
{
'AttributeName': 'name',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
)
def delete_db_database(db_name, region):
dynamodb = boto3.resource('dynamodb', region_name=region)
table = dynamodb.Table(db_name)
table.delete()
def create_tf_state_store(name, region):
my_config = Config(region_name = region)
s3 = boto3.client('s3', config=my_config)
response = s3.create_bucket(Bucket=name, CreateBucketConfiguration={'LocationConstraint': region})
client = boto3.client('dynamodb', config=my_config)
response = client.create_table(
TableName=name,
KeySchema=[
{
'AttributeName': 'LockID',
'KeyType': 'HASH' # Partition key
}
],
AttributeDefinitions=[
{
'AttributeName': 'LockID',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
)
def delete_tf_state_store(region, name):
s3 = boto3.resource('s3', region_name=region)
bucket = s3.Bucket(name)
bucket.objects.all().delete()
bucket.delete()
dynamodb = boto3.resource('dynamodb', region_name=region)
table = dynamodb.Table(name)
table.delete()
def get_secret(secret_name):
region_name = "eu-central-1"
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
get_secret_value_response = client.get_secret_value(SecretId=secret_name)
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
secret_obj = json.loads(secret)
return secret_obj[secret_name]
@@ -1,16 +0,0 @@
import git
import os
import logging
from helpers import aws_service
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
def clone_attack_range_project():
LOGGER.info(f"Clone Attack Range Project")
O_AUTH_TOKEN_GITHUB = aws_service.get_secret("github_token")
repo_obj = git.Repo.clone_from('https://' + O_AUTH_TOKEN_GITHUB + ':x-oauth-basic@github.com/splunk/attack_range', "attack_range", branch='develop')
return repo_obj
@@ -1,79 +0,0 @@
ansible==4.2.0
ansible-runner==1.4.7
apipkg==1.5
aspy.yaml==1.3.0
atomicwrites==1.4.0
attackcti==0.3.4.3
attrs==20.3.0
azure-common==1.1.25
azure-core==1.8.2
azure-identity==1.4.1
azure-mgmt-compute==17.0.0
azure-mgmt-core==1.2.1
azure-mgmt-network==16.0.0
azure-mgmt-resource==15.0.0
bcrypt==3.2.0
boto3==1.17.30
botocore==1.20.30
certifi==2020.12.5
cffi==1.14.5
cfgv==2.0.1
chardet==4.0.0
configparser==5.0.2
contextlib2==0.6.0.post1
Deprecated==1.2.12
dnspython==2.1.0
docutils==0.16
execnet==1.8.0
gitdb==4.0.5
GitPython==3.1.14
identify==2.1.3
idna==2.8
importlib-metadata==3.7.3
Jinja2==2.11.3
jmespath==0.10.0
lockfile==0.12.2
MarkupSafe==1.1.1
mock==4.0.3
more-itertools==8.7.0
mysql-connector-python==8.0.23
nodeenv==1.3.4
ntlm-auth==1.5.0
packaging==20.9
path==15.1.2
path.py==12.5.0
pexpect==4.8.0
pluggy==0.13.1
pre-commit==2.11.1
protobuf==3.15.6
psutil==5.8.0
ptyprocess==0.7.0
py==1.10.0
pycparser==2.20
PyGithub==1.54.1
PyInquirer==1.0.3
PyJWT<2.0.1
PyNaCl==1.4.0
pyparsing==2.4.7
pytest==6.2.2
python-daemon==2.3.0
python-dateutil==2.8.1
python-terraform==0.10.1
pywinrm==0.4.1
PyYAML==5.4.1
requests==2.25.1
requests-ntlm==1.1.0
s3transfer==0.3.4
six==1.13.0
smmap==3.0.5
splunk-sdk==1.6.15
tabulate==0.8.9
termcolor==1.1.0
toml==0.10.2
urllib3==1.26.5
virtualenv==20.4.3
wcwidth==0.2.5
wget==3.2
wrapt==1.12.1
xmltodict==0.12.0
zipp==3.4.1
@@ -1,14 +0,0 @@
FROM ubuntu:18.04
RUN apt-get update
RUN DEBIAN_FRONTEND="noninteractive" apt-get -y install tzdata
RUN apt-get install -y python3-dev git python-dev unzip python3-pip awscli
RUN apt-get install -y python-gitdb
RUN apt-get install -y wget unzip
ADD . /app
WORKDIR /app
RUN pip3 install -r requirements.txt
ENTRYPOINT ["python3", "labeled_data.py"]
@@ -1,6 +0,0 @@
---
- hosts: all
gather_facts: False
roles:
- attack_replay
@@ -1,23 +0,0 @@
---
- name: Upload replay
copy:
src: ../../../{{ folder_name }}/{{ out }}
dest: /tmp/{{ out }}
- name: Call oneshot import
uri:
url: https://localhost:8089/services/data/inputs/oneshot
validate_certs: no
method: POST
user: admin
password: "{{ splunk_password }}"
force_basic_auth: yes
body_format: form-urlencoded
body:
name: /tmp/{{ out }}
sourcetype: "{{ sourcetype }}"
rename-source: "{{ source }}"
index: "{{ index }}"
status_code: 201
@@ -1,21 +0,0 @@
---
- name: Delete ESCU APP
file:
state: absent
path: "/opt/splunk/etc/apps/DA-ESS-ContentUpdate"
become: yes
- name: Upload ESCU APP
copy:
src: ../../../{{ security_content_path }}/dist/escu/
dest: "/opt/splunk/etc/apps/DA-ESS-ContentUpdate"
owner: splunk
group: splunk
become: yes
- name: restart splunk
service:
name: splunkd
state: restarted
become: yes
@@ -1,4 +0,0 @@
- hosts: all
gather_facts: False
roles:
- update_escu
@@ -1,40 +0,0 @@
import sys
import argparse
import shutil
import os
from modules.github_service import GithubService
from modules import aws_service, testing_service
DT_ATTACK_RANGE_STATE_STORE = "dt-attack-range-tf-state-store"
DT_ATTACK_RANGE_STATE = "dt-attack-range-state"
REGION = "eu-central-1"
NAME = "detection-testing-attack-range"
def main(args):
github_service = GithubService('develop')
test_files = github_service.get_test_files()
for test_file in test_files:
print(test_file)
dt_ar = aws_service.get_ar_information_from_dynamo_db(REGION, DT_ATTACK_RANGE_STATE)
splunk_instance = aws_service.get_splunk_instance(REGION, dt_ar['ssh_key_name'])
splunk_ip = splunk_instance['NetworkInterfaces'][0]['Association']['PublicIp']
splunk_password = dt_ar['password']
ssh_key_name = dt_ar['ssh_key_name']
private_key = dt_ar['private_key']
testing_service.prepare_detection_testing(ssh_key_name, private_key, splunk_ip, splunk_password)
testing_service.test_detections(ssh_key_name, private_key, splunk_ip, splunk_password, test_files)
# Get Password, private_key and key_name from DynamoDB
# For loop
# Detection Test
if __name__ == "__main__":
main(sys.argv[1:])
@@ -1,131 +0,0 @@
import json
from datetime import datetime
from datetime import timedelta
import fileinput
import os
import re
import io
class DataManipulation:
def manipulate_timestamp(self, file_path, sourcetype, source):
print('Updating timestamps in attack_data before replaying')
if sourcetype == 'aws:cloudtrail':
self.manipulate_timestamp_cloudtrail(file_path)
if source == 'WinEventLog:System' or source == 'WinEventLog:Security':
self.manipulate_timestamp_windows_event_log_raw(file_path)
if source == 'exchange':
self.manipulate_timestamp_exchange_logs(file_path)
def manipulate_timestamp_exchange_logs(self, file_path):
path = os.path.join(os.path.dirname(__file__), '../' + file_path)
path = path.replace('modules/../','')
f = io.open(path, "r", encoding="utf-8")
first_line = f.readline()
d = json.loads(first_line)
latest_event = datetime.strptime(d["CreationTime"],"%Y-%m-%dT%H:%M:%S")
now = datetime.now()
now = now.strftime("%Y-%m-%dT%H:%M:%S")
now = datetime.strptime(now,"%Y-%m-%dT%H:%M:%S")
difference = now - latest_event
f.close()
for line in fileinput.input(path, inplace=True):
d = json.loads(line)
original_time = datetime.strptime(d["CreationTime"],"%Y-%m-%dT%H:%M:%S")
new_time = (difference + original_time)
original_time = original_time.strftime("%Y-%m-%dT%H:%M:%S")
new_time = new_time.strftime("%Y-%m-%dT%H:%M:%S")
print (line.replace(original_time, new_time),end ='')
def manipulate_timestamp_windows_event_log_raw(self, file_path):
path = os.path.join(os.path.dirname(__file__), '../' + file_path)
path = path.replace('modules/../','')
f = io.open(path, "r", encoding="utf-8")
self.now = datetime.now()
self.now = self.now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
self.now = datetime.strptime(self.now,"%Y-%m-%dT%H:%M:%S.%fZ")
# read raw logs
regex = r'\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2} [AP]M'
data = f.read()
lst_matches = re.findall(regex, data)
if len(lst_matches) > 0:
latest_event = datetime.strptime(lst_matches[-1],"%m/%d/%Y %I:%M:%S %p")
self.difference = self.now - latest_event
f.close()
result = re.sub(regex, self.replacement_function, data)
with io.open(path, "w+", encoding='utf8') as f:
f.write(result)
else:
f.close()
return
def replacement_function(self, match):
try:
event_time = datetime.strptime(match.group(),"%m/%d/%Y %I:%M:%S %p")
new_time = self.difference + event_time
return new_time.strftime("%m/%d/%Y %I:%M:%S %p")
except Exception as e:
self.logger.error("Error in timestamp replacement occured: " + str(e))
return match.group()
def manipulate_timestamp_cloudtrail(self, file_path):
path = os.path.join(os.path.dirname(__file__), '../' + file_path)
path = path.replace('modules/../','')
f = io.open(path, "r", encoding="utf-8")
try:
first_line = f.readline()
d = json.loads(first_line)
latest_event = datetime.strptime(d["eventTime"],"%Y-%m-%dT%H:%M:%S.%fZ")
now = datetime.now()
now = now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
now = datetime.strptime(now,"%Y-%m-%dT%H:%M:%S.%fZ")
except ValueError:
first_line = f.readline()
d = json.loads(first_line)
latest_event = datetime.strptime(d["eventTime"],"%Y-%m-%dT%H:%M:%SZ")
now = datetime.now()
now = now.strftime("%Y-%m-%dT%H:%M:%SZ")
now = datetime.strptime(now,"%Y-%m-%dT%H:%M:%SZ")
difference = now - latest_event
f.close()
for line in fileinput.input(path, inplace=True):
try:
d = json.loads(line)
original_time = datetime.strptime(d["eventTime"],"%Y-%m-%dT%H:%M:%S.%fZ")
new_time = (difference + original_time)
original_time = original_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
new_time = new_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
print (line.replace(original_time, new_time),end ='')
except ValueError:
d = json.loads(line)
original_time = datetime.strptime(d["eventTime"],"%Y-%m-%dT%H:%M:%SZ")
new_time = (difference + original_time)
original_time = original_time.strftime("%Y-%m-%dT%H:%M:%SZ")
new_time = new_time.strftime("%Y-%m-%dT%H:%M:%SZ")
print (line.replace(original_time, new_time),end ='')
@@ -1,84 +0,0 @@
import boto3
def get_ar_information_from_dynamo_db(region, db_name):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table(db_name)
response = table.get_item(
Key={
'name': 'detection-testing-attack-range'
}
)
if 'Item' in response:
return response['Item']
else:
return {}
def get_splunk_instance(region, key_name):
client = boto3.client('ec2', region_name=region)
response = client.describe_instances(
Filters=[
{
'Name': "key-name",
'Values': [key_name]
}
]
)
instances = []
for reservation in response['Reservations']:
for instance in reservation['Instances']:
if instance['State']['Name']!='terminated':
if len(instance['Tags']) > 0:
tag = instance['Tags'][0]['Value']
if key_name in tag:
return instance
return {}
def add_detection_results_in_dynamo_db(region, uuid, uuid_test, detection, detection_path, time):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table("dt-results")
response = table.put_item(Item= {
'uuid': uuid,
'uuid_test': uuid_test,
'detection': detection,
'detection_path': detection_path,
'time': time,
'status': 'running'
})
def update_detection_results_in_dynamo_db(region, uuid, result):
resource = boto3.resource('dynamodb', region_name=region)
table = resource.Table("dt-results")
response = table.update_item(
Key={
'uuid': uuid
},
UpdateExpression="set #ts=:s",
ExpressionAttributeValues={
':s': 'done'
},
ExpressionAttributeNames={
"#ts": "status"
},
ReturnValues="UPDATED_NEW"
)
response = table.update_item(
Key={
'uuid': uuid
},
UpdateExpression="set #ts=:s",
ExpressionAttributeValues={
':s': result
},
ExpressionAttributeNames={
"#ts": "result"
},
ReturnValues="UPDATED_NEW"
)
@@ -1,77 +0,0 @@
import git
import os
import logging
import glob
import subprocess
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
SECURITY_CONTENT_URL = "https://github.com/splunk/security_content"
class GithubService:
def __init__(self, security_content_branch):
self.security_content_branch = security_content_branch
self.security_content_repo_obj = self.clone_project(SECURITY_CONTENT_URL, f"security_content", f"develop")
self.security_content_repo_obj.git.checkout(security_content_branch)
def clone_project(self, url, project, branch):
LOGGER.info(f"Clone Security Content Project")
repo_obj = git.Repo.clone_from(url, project, branch=branch)
return repo_obj
def get_test_files(self):
path = 'security_content/tests/endpoint'
test_files = []
for root, directories, files in os.walk(path, topdown=False):
for name in files:
file_path = os.path.join(root, name)
if not os.path.basename(file_path).startswith('ssa') and os.path.basename(file_path).endswith('.yml'):
test_files.append(file_path)
# for testing
#test_files = ['security_content/tests/endpoint/certutil_exe_certificate_extraction.test.yml']
test_files.remove('security_content/tests/endpoint/winword_spawning_windows_script_host.test.yml')
test_files.remove('security_content/tests/endpoint/winword_spawning_powershell.test.yml')
test_files.remove('security_content/tests/endpoint/winword_spawning_cmd.test.yml')
# test_files.remove('security_content/tests/endpoint/bitsadmin_download_file.test.yml')
# test_files.remove('security_content/tests/endpoint/bcdedit_failure_recovery_modification.test.yml')
return sorted(test_files, reverse=True)[61:]
# def get_changed_test_files(self):
# branch1 = self.security_content_branch
# branch2 = 'develop'
# g = git.Git('security_content')
# changed_test_files = []
# if branch1 != 'develop':
# differ = g.diff('--name-status', branch2 + '...' + branch1)
# changed_files = differ.splitlines()
# for file_path in changed_files:
# # added or changed test files
# if file_path.startswith('A') or file_path.startswith('M'):
# if 'tests' in file_path:
# if not os.path.basename(file_path).startswith('ssa') and os.path.basename(file_path).endswith('.test.yml'):
# if file_path not in changed_test_files:
# changed_test_files.append(file_path)
# # changed detections
# if 'detections' in file_path:
# if not os.path.basename(file_path).startswith('ssa') and os.path.basename(file_path).endswith('.yml'):
# file_path_base = os.path.splitext(file_path)[0].replace('detections', 'tests') + '.test'
# file_path_new = file_path_base + '.yml'
# if file_path_new not in changed_test_files:
# changed_test_files.append(file_path_new)
# return changed_test_files
@@ -1,176 +0,0 @@
import sys
from time import sleep
import splunklib.results as results
import splunklib.client as client
import splunklib.results as results
import requests
def test_baseline_search(splunk_host, splunk_password, search, pass_condition, baseline_name, baseline_file, earliest_time, latest_time):
try:
service = client.connect(
host=splunk_host,
port=8089,
username='admin',
password=splunk_password
)
except Exception as e:
print("Unable to connect to Splunk instance: " + str(e))
return 1, {}
# search and replace \\ with \\\
# search = search.replace('\\','\\\\')
if search.startswith('|'):
search = search
else:
search = 'search ' + search
kwargs = {"exec_mode": "blocking",
"dispatch.earliest_time": earliest_time,
"dispatch.latest_time": latest_time}
splunk_search = search + ' ' + pass_condition
try:
job = service.jobs.create(splunk_search, **kwargs)
except Exception as e:
print("Unable to execute baseline: " + str(e))
return 1, {}
test_results = dict()
test_results['diskUsage'] = job['diskUsage']
test_results['runDuration'] = job['runDuration']
test_results['baseline_name'] = baseline_name
test_results['baseline_file'] = baseline_file
test_results['scanCount'] = job['scanCount']
if int(job['resultCount']) != 1:
print("Test failed for baseline: " + baseline_name)
test_results['error'] = True
return test_results
else:
print("Test successful for baseline: " + baseline_name)
test_results['error'] = False
return test_results
def run_modified_splunk_search(splunk_host, splunk_password, search1, search2, detection_name, detection_file, earliest_time, latest_time):
try:
service = client.connect(
host=splunk_host,
port=8089,
username='admin',
password=splunk_password
)
except Exception as e:
print("Unable to connect to Splunk instance: " + str(e))
return 1, {}
if not search1.startswith('|'):
search1 = 'search ' + search1
if not search2.startswith('|'):
search2 = 'search ' + search2
kwargs = {"dispatch.earliest_time": "-1d",
"dispatch.latest_time": "now"}
try:
job = service.jobs.export(search1, **kwargs)
except Exception as e:
print("Unable to execute detection: " + str(e))
return 1, {}
reader = results.ResultsReader(job)
results_search_malicious = []
for result in reader:
if isinstance(result, dict):
results_search_malicious.append(result)
try:
job = service.jobs.export(search2, **kwargs)
except Exception as e:
print("Unable to execute detection: " + str(e))
return 1, {}
reader = results.ResultsReader(job)
results_search_not_malicious = []
for result in reader:
if isinstance(result, dict):
results_search_not_malicious.append(result)
return results_search_malicious, results_search_not_malicious
def test_detection_search(splunk_host, splunk_password, search, pass_condition, detection_name, detection_file, earliest_time, latest_time):
try:
service = client.connect(
host=splunk_host,
port=8089,
username='admin',
password=splunk_password
)
except Exception as e:
print("Unable to connect to Splunk instance: " + str(e))
return 1, {}
# search and replace \\ with \\\
# search = search.replace('\\','\\\\')
if search.startswith('|'):
search = search
else:
search = 'search ' + search
kwargs = {"exec_mode": "blocking",
"dispatch.earliest_time": "-1d",
"dispatch.latest_time": "now"}
splunk_search = search + ' ' + pass_condition
try:
job = service.jobs.create(splunk_search, **kwargs)
except Exception as e:
print("Unable to execute detection: " + str(e))
return 1, {}
test_results = dict()
test_results['diskUsage'] = job['diskUsage']
test_results['runDuration'] = job['runDuration']
test_results['detection_name'] = detection_name
test_results['detection_file'] = detection_file
test_results['scanCount'] = job['scanCount']
if int(job['resultCount']) != 1:
print("Test failed for detection: " + detection_name)
test_results['error'] = True
return test_results
else:
print("Test successful for detection: " + detection_name)
test_results['error'] = False
return test_results
def delete_attack_data(splunk_host, splunk_password):
try:
service = client.connect(
host=splunk_host,
port=8089,
username='admin',
password=splunk_password
)
except Exception as e:
print("Unable to connect to Splunk instance: " + str(e))
return 1, {}
splunk_search = 'search index=test* | delete'
kwargs = {"exec_mode": "blocking",
"dispatch.earliest_time": "-360d",
"dispatch.latest_time": "now"}
try:
job = service.jobs.create(splunk_search, **kwargs)
except Exception as e:
print("Unable to execute search: " + str(e))
return 1, {}
@@ -1,279 +0,0 @@
import ansible_runner
import yaml
import uuid
import sys
import os
import time
import requests
import re
import csv
import boto3
from botocore.exceptions import ClientError
from modules.DataManipulation import DataManipulation
from modules import splunk_sdk, aws_service
TSTATS_SEARCH = """| tstats count as count values(Processes.action) as action,
values(Processes.cpu_load_percent) as cpu_load_percent,
values(Processes.dest) as dest,
values(Processes.mem_used) as mem_used,
values(Processes.os) as os,
values(Processes.parent_process) as parent_process,
values(Processes.parent_process_exec) as parent_process_exec,
values(Processes.parent_process_id) as parent_process_id,
values(Processes.parent_process_guid) as parent_process_guid,
values(Processes.parent_process_name) as parent_process_name,
values(Processes.parent_process_path) as parent_process_path,
values(Processes.process) as process,
values(Processes.process_current_directory) as process_current_directory,
values(Processes.process_exec) as process_exec,
values(Processes.process_hash) as process_hash,
values(Processes.process_guid) as process_guid,
values(Processes.process_id) as process_id,
values(Processes.process_integrity_level) as process_integrity_level,
values(Processes.process_name) as process_name,
values(Processes.process_path) as process_path,
values(Processes.tag) as tag,
values(Processes.user) as user,
values(Processes.user_id) as user_id,
values(Processes.vendor_product) as vendor_product,
values(host) as host,
values(source) as source,
values(sourcetype) as sourcetype
from datamodel=Endpoint.Processes
where
"""
TSTATS_INVERSE_SEARCH = """| tstats count as count values(Processes.action) as action,
values(Processes.cpu_load_percent) as cpu_load_percent,
values(Processes.dest) as dest,
values(Processes.mem_used) as mem_used,
values(Processes.os) as os,
values(Processes.parent_process) as parent_process,
values(Processes.parent_process_exec) as parent_process_exec,
values(Processes.parent_process_id) as parent_process_id,
values(Processes.parent_process_guid) as parent_process_guid,
values(Processes.parent_process_name) as parent_process_name,
values(Processes.parent_process_path) as parent_process_path,
values(Processes.process) as process,
values(Processes.process_current_directory) as process_current_directory,
values(Processes.process_exec) as process_exec,
values(Processes.process_hash) as process_hash,
values(Processes.process_guid) as process_guid,
values(Processes.process_id) as process_id,
values(Processes.process_integrity_level) as process_integrity_level,
values(Processes.process_name) as process_name,
values(Processes.process_path) as process_path,
values(Processes.tag) as tag,
values(Processes.user) as user,
values(Processes.user_id) as user_id,
values(Processes.vendor_product) as vendor_product,
values(host) as host,
values(source) as source,
values(sourcetype) as sourcetype
from datamodel=Endpoint.Processes
where NOT(_replace1_)
by _replace2_
"""
def prepare_detection_testing(ssh_key_name, private_key, splunk_ip, splunk_password):
with open(ssh_key_name, 'w') as file :
file.write(private_key)
os.chmod(ssh_key_name, 0o600)
sys.path.append(os.path.join(os.getcwd(),'security_content/bin'))
try:
module = __import__('generate')
results = module.main(REPO_PATH = 'security_content' , OUTPUT_PATH = 'security_content/dist/escu', PRODUCT = 'ESCU', VERBOSE = 'False' )
except Exception as e:
print('Error: ' + str(e))
update_ESCU_app(splunk_ip, ssh_key_name, splunk_password)
def test_detections(ssh_key_name, private_key, splunk_ip, splunk_password, test_files):
test_index = 1
for test_file in test_files:
test_detection(ssh_key_name, private_key, splunk_ip, splunk_password, test_file, test_index)
if test_index == 10:
test_index = 1
else:
test_index = test_index + 1
# delete test data
splunk_sdk.delete_attack_data(splunk_ip, splunk_password)
def test_detection(ssh_key_name, private_key, splunk_ip, splunk_password, test_file, test_index):
test_file_obj = load_file(test_file)
if not test_file_obj:
return
print("\nAnalysis of detection: " + test_file_obj['tests'][0]['name'])
test = test_file_obj['tests'][0]
detection_file_name = test['file']
detection = load_file(os.path.join(os.path.dirname(__file__), '../security_content/detections', detection_file_name))
print("Modify detection")
detection_search_1, detection_search_2 = modify_detection(detection['search'])
if detection_search_1 == "ERROR" or detection_search_2 == "ERROR":
print("ERROR: Detection doesn't use Endpoint.Processes Data Model")
return
epoch_time = str(int(time.time()))
folder_name = "attack_data_" + epoch_time
os.mkdir(folder_name)
for attack_data in test_file_obj['tests'][0]['attack_data']:
url = attack_data['data']
r = requests.get(url, allow_redirects=True)
open(folder_name + '/' + attack_data['file_name'], 'wb').write(r.content)
# Update timestamps before replay
if 'update_timestamp' in attack_data:
if attack_data['update_timestamp'] == True:
data_manipulation = DataManipulation()
data_manipulation.manipulate_timestamp(folder_name + '/' + attack_data['file_name'], attack_data['sourcetype'], attack_data['source'])
print("Replay Attack Data")
replay_attack_dataset(splunk_ip, splunk_password, ssh_key_name, folder_name, 'test' + str(test_index), attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
time.sleep(60)
# result_test = {}
# if 'baselines' in test:
# results_baselines = []
# for baseline_obj in test['baselines']:
# baseline_file_name = baseline_obj['file']
# baseline = load_file(os.path.join(os.path.dirname(__file__), '../security_content', baseline_file_name))
# result_obj = dict()
# result_obj['baseline'] = baseline_obj['name']
# result_obj['baseline_file'] = baseline_obj['file']
# result = splunk_sdk.test_baseline_search(splunk_ip, splunk_password, baseline['search'], baseline_obj['pass_condition'], baseline['name'], baseline_obj['file'], baseline_obj['earliest_time'], baseline_obj['latest_time'])
# result_test['baselines_result'] = results_baselines
# result_detection = splunk_sdk.test_detection_search(splunk_ip, splunk_password, detection['search'], test['pass_condition'], detection['name'], test['file'], test['earliest_time'], test['latest_time'])
# for testing
print("Run Splunk Search")
results_1, results_2 = splunk_sdk.run_modified_splunk_search(splunk_ip, splunk_password, detection_search_1, detection_search_2, detection['name'], test['file'], test['earliest_time'], test['latest_time'])
if len(results_1) > 0 and len(results_2) > 0:
print("Write Results to csv")
try:
with open('test.csv', 'w') as csvfile:
results_1[0]['malicious'] = "yes"
field_names = results_1[0].keys()
writer = csv.DictWriter(csvfile, fieldnames=field_names)
writer.writeheader()
for data in results_1:
data = {k: str(v).encode("utf-8").decode() for k,v in data.items()}
data['malicious'] = 'yes'
writer.writerow(data)
for data in results_2:
data = {k: str(v).encode("utf-8").decode() for k,v in data.items()}
data['malicious'] = 'no'
writer.writerow(data)
detection_name = os.path.splitext(os.path.basename(detection_file_name))[0]
# Upload the file
print("S3 upload results")
s3_client = boto3.client('s3')
try:
response = s3_client.upload_file('test.csv', 'security-content-labeled-data', 'endpoint/' + detection_name + '/' + detection_name + '.csv')
except ClientError as e:
print(e)
except Exception as e:
print(e)
else:
print("ERROR: Detection didn't return results")
# result_detection['detection_name'] = test['name']
# result_detection['detection_file'] = test['file']
# result_test['detection_result'] = result_detection
def load_file(file_path):
try:
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
print("ERROR: reading {0}".format(file_path))
return False
except Exception as e:
print("ERROR: reading {0}".format(file_path))
return False
return file
def update_ESCU_app(splunk_ip, ssh_key_name, splunk_password):
print("Update ESCU App. This can take some time")
ansible_vars = {}
ansible_vars['ansible_user'] = 'ubuntu'
ansible_vars['ansible_ssh_private_key_file'] = ssh_key_name
ansible_vars['splunk_password'] = splunk_password
ansible_vars['security_content_path'] = 'security_content'
cmdline = "-i %s, -u ubuntu" % (splunk_ip)
runner = ansible_runner.run(private_data_dir=os.path.join(os.path.dirname(__file__), '../'),
cmdline=cmdline,
roles_path=os.path.join(os.path.dirname(__file__), '../ansible/roles'),
playbook=os.path.join(os.path.dirname(__file__), '../ansible/update_escu.yml'),
extravars=ansible_vars)
def replay_attack_dataset(splunk_ip, splunk_password, ssh_key_name, folder_name, index, sourcetype, source, out):
ansible_vars = {}
ansible_vars['folder_name'] = folder_name
ansible_vars['ansible_user'] = 'ubuntu'
ansible_vars['ansible_ssh_private_key_file'] = ssh_key_name
ansible_vars['splunk_password'] = splunk_password
ansible_vars['out'] = out
ansible_vars['sourcetype'] = sourcetype
ansible_vars['source'] = source
ansible_vars['index'] = index
cmdline = "-i %s, -u ubuntu" % (splunk_ip)
runner = ansible_runner.run(private_data_dir=os.path.join(os.path.dirname(__file__), '../'),
cmdline=cmdline,
roles_path=os.path.join(os.path.dirname(__file__), '../ansible/roles'),
playbook=os.path.join(os.path.dirname(__file__), '../ansible/attack_replay.yml'),
extravars=ansible_vars)
def modify_detection(splunk_search):
if splunk_search.startswith('| tstats'):
if "from datamodel=Endpoint.Processes" in splunk_search:
regex1 = r'where ([^\|]*)'
a = re.search(regex1, splunk_search)
search1 = ""
if a:
search1 = str(TSTATS_SEARCH + a.group(1))
else:
search1 = "ERROR"
regex2 = r'where (.*)by ([^\|]+)'
b = re.search(regex2, splunk_search)
search2 = ""
if b:
search2 = TSTATS_INVERSE_SEARCH.replace("_replace1_", b.group(1))
search2 = search2.replace("_replace2_", b.group(2))
else:
search2 = "ERROR"
return search1, search2
return "ERROR", "ERROR"
@@ -1,10 +0,0 @@
ansible==4.2.0
ansible-runner==1.4.4
attackcti==0.3.4.3
boto3==1.11.0
GitPython==3.1.14
Jinja2==3.0.0
PyYAML==5.4
requests==2.25.1
six==1.16.0
splunk-sdk==1.6.12
@@ -1,90 +0,0 @@
import uuid
import sys
import boto3
import argparse
import time
def main(args):
parser = argparse.ArgumentParser(description="Detection Testing Execution")
parser.add_argument("-b", "--branch", required=True, help="security content branch")
parser.add_argument("-pr", "--pr-number", required=False, help="Pull Request Number")
args = parser.parse_args()
branch = args.branch
pr_number = args.pr_number
# vars
max_waiting_time = 7200
current_waiting_time = 0
# create uuid
uuid_test = str(uuid.uuid4())
# start aws batch job
client = boto3.client("batch", region_name="eu-central-1")
if pr_number:
response = client.submit_job(
jobName='detection_testing_' + branch,
jobQueue='detection_testing_execution_queue',
jobDefinition='detection_testing_execution:2',
containerOverrides={
'command': ['-b', branch, '-u', uuid_test, '-pr', pr_number]
}
)
else:
response = client.submit_job(
jobName='detection_testing_' + branch,
jobQueue='detection_testing_execution_queue',
jobDefinition='detection_testing_execution:2',
containerOverrides={
'command': ['-b', branch, '-u', uuid_test]
}
)
while max_waiting_time > current_waiting_time:
dynamodb = boto3.client('dynamodb', region_name="eu-central-1")
response = dynamodb.query(
TableName='dt-results',
IndexName='uuid_test-index',
KeyConditionExpression='uuid_test = :uuid_test',
ExpressionAttributeValues={
':uuid_test': {'S': uuid_test}
}
)
test_done = True
for item in response['Items']:
if item['status']['S'] == 'nothing to test':
print('No new or changed detections to test.')
sys.exit(0)
if item['status']['S'] == 'running':
test_done = False
if len(response['Items']) == 0 or (not test_done):
time.sleep(60)
current_waiting_time = current_waiting_time + 60
else:
test_passed = True
for item in response['Items']:
if item['result']['S'] == 'failed':
test_passed = False
print('Test failed for detection: ' + item['detection']['S'] + ', ' + item['detection_path']['S'])
else:
print('Test passed for detection: ' + item['detection']['S'] + ', ' + item['detection_path']['S'])
sys.exit(not test_passed)
sys.exit(1)
if __name__ == "__main__":
main(sys.argv[1:])
@@ -1 +0,0 @@
boto3==1.18.35
-2
View File
@@ -1,2 +0,0 @@
[default]
region = eu-central-1
@@ -1,237 +0,0 @@
import os
from os import path
import sys
import argparse
import git
from shutil import copyfile
from shutil import which
import subprocess
import boto3
from random import randrange
import yaml
from github import Github
from jinja2 import Environment, FileSystemLoader
import base64
from botocore.exceptions import ClientError
import json
from datetime import datetime
import subprocess
import time
def main(args):
parser = argparse.ArgumentParser(description="detection testing service based on Attack Range.")
parser.add_argument("-tfn", "--test_file_name", required=True,
help="specify the test file name located in security content repository")
parser.add_argument("-arr", "--attack_range_repo", required=False, default="splunk/attack_range",
help="specify the url of the atack range repository")
parser.add_argument("-arb", "--attack_range_branch", required=False, default="develop",
help="specify the atack range branch")
parser.add_argument("-scr", "--security_content_repo", required=False, default="splunk/security_content",
help="specify the url of the security content repository")
parser.add_argument("-scb", "--security_content_branch", required=False, default="develop",
help="specify the security content branch")
parser.add_argument("-gt", "--github_token", required=False,
help="specify the github token for the PR")
parser.add_argument("-smk", "--secrets_manager_key", required=False, default="github_token",
help="specify the key in AWS secrets manager for your github token")
parser.add_argument("-s3b", "--s3_bucket", required=False, default="attack-range-automated-testing",
help="S3 bucket to store the test data")
args = parser.parse_args()
test_file_name = args.test_file_name
attack_range_repo = args.attack_range_repo
attack_range_branch = args.attack_range_branch
security_content_repo = args.security_content_repo
security_content_branch = args.security_content_branch
github_token = args.github_token
secrets_manager_key = args.secrets_manager_key
s3_bucket = args.s3_bucket
test_obj = {}
test_obj['technique'] = 'T1003.001'
detection_obj = {}
detection_obj['detection'] = 'Access LSASS Memory for Dump Creation'
test_obj['results'] = [detection_obj]
# get github token
if github_token:
O_AUTH_TOKEN_GITHUB = github_token
else:
O_AUTH_TOKEN_GITHUB = get_secret(secrets_manager_key)
# clone repositories
git.Repo.clone_from('https://github.com/' + attack_range_repo, "attack_range", branch=attack_range_branch)
security_content_repo_obj = git.Repo.clone_from('https://' + O_AUTH_TOKEN_GITHUB + ':x-oauth-basic@github.com/' + security_content_repo, "security_content", branch=security_content_branch)
sys.path.append(os.path.join(os.getcwd(),'attack_range'))
copyfile('attack_range/attack_range.conf.template', 'attack_range/attack_range.conf')
epoch_time = str(int(time.time()))
ssh_key_name = 'ds-key-pair-' + epoch_time
# create ssh keys
ec2 = boto3.client('ec2')
response = ec2.create_key_pair(KeyName=ssh_key_name)
with open(ssh_key_name, "w") as ssh_key:
ssh_key.write(response['KeyMaterial'])
os.chmod(ssh_key_name, 0o600)
private_key_path = str(os.getcwd() + "/" + ssh_key_name)
# build new version of ESCU
sys.path.append(os.path.join(os.getcwd(),'security_content/bin'))
try:
module = __import__('generate')
results = module.main(REPO_PATH = 'security_content' , OUTPUT_PATH = 'security_content/dist/escu', PRODUCT = 'ESCU', VERBOSE = 'True' )
except Exception as e:
print('Error: ' + str(e))
with open('attack_range/attack_range.conf', 'r') as file :
filedata = file.read()
filedata = filedata.replace('attack_range_password = Pl3ase-k1Ll-me:p', 'attack_range_password = I-l1ke-Attack-Range!')
filedata = filedata.replace('windows_domain_controller = 1', 'windows_domain_controller = 0')
filedata = filedata.replace('windows_server_join_domain = 1', 'windows_server_join_domain = 0')
filedata = filedata.replace('region = us-west-2', 'region = eu-central-1')
filedata = filedata.replace('key_name = attack-range-key-pair', 'key_name = ' + ssh_key_name)
filedata = filedata.replace('private_key_path = ~/.ssh/id_rsa', 'private_key_path = ' + private_key_path)
filedata = filedata.replace('update_escu_app = 0', 'update_escu_app = 1')
with open('attack_range/attack_range.conf', 'w') as file:
file.write(filedata)
module = __import__('attack_range')
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'test', '--test_file', 'security_content/tests/' + test_file_name, '--test_build_destroy']
execution_error = False
try:
results = module.main(module.sys.argv)
except Exception as e:
print('Error: ' + str(e))
module.sys.argv = ['attack_range', '--config', 'attack_range/attack_range.conf', 'destroy']
module.main(module.sys.argv)
execution_error = True
# delete ssh key
response = ec2.delete_key_pair(KeyName=ssh_key_name)
# read_test_file
test_file = load_file('security_content/tests/' + test_file_name)
# check if was succesful
if not execution_error:
# Create GitHub PR security content
if security_content_branch == 'develop':
branch_name = "automated_detection_testing_" + epoch_time
security_content_repo_obj.git.checkout(security_content_branch, b=branch_name)
else:
branch_name = security_content_branch
security_content_repo_obj.git.checkout(security_content_branch)
counter = 0
for test in results:
if not test['detection_result']['error']:
file_path = 'security_content/detections/' + test['detection_result']['detection_file']
detection_obj = load_file(file_path)
detection_obj['tags']['automated_detection_testing'] = 'passed'
if 'attack_data' in test_file['tests'][counter]:
datasets = []
for dataset in test_file['tests'][counter]['attack_data']:
datasets.append(dataset['data'])
detection_obj['tags']['dataset'] = datasets
with open(file_path, 'w') as f:
yaml.dump(detection_obj, f, sort_keys=False, allow_unicode=True)
changed_file_path = 'detections/' + test['detection_result']['detection_file']
security_content_repo_obj.index.add([changed_file_path])
security_content_repo_obj.index.commit('Added detection testing service results in' + test['detection_result']['detection_name'])
counter = counter + 1
j2_env = Environment(loader=FileSystemLoader('templates'),trim_blocks=True) # nosemgrep
template = j2_env.get_template('PR_template.j2')
body = template.render(results=results)
security_content_repo_obj.config_writer().set_value("user", "name", "Detection Testing Service").release()
security_content_repo_obj.config_writer().set_value("user", "email", "research@splunk.com").release()
if not security_content_branch == 'develop':
security_content_repo_obj.remotes.origin.pull()
security_content_repo_obj.git.push('--set-upstream', 'origin', branch_name)
g = Github(O_AUTH_TOKEN_GITHUB)
repo = g.get_repo("splunk/security_content")
pull_requests = repo.get_pulls(state='open', sort='created', head=branch_name)
for pr in pull_requests:
if pr.head.label == str('splunk:' + branch_name):
pr.create_issue_comment(body)
exit(0)
pr = repo.create_pull(title="Automated Detection Testing PR " + branch_name, body=body, head=branch_name, base="develop")
def load_file(file_path):
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
print(exc)
sys.exit("ERROR: reading {0}".format(file_path))
return file
def get_secret(secret_name):
region_name = "eu-central-1"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
secret_obj = json.loads(secret)
return secret_obj['github_token']
if __name__ == "__main__":
main(sys.argv[1:])
@@ -1,83 +0,0 @@
ansible==4.7.0
ansible-runner==2.0.2
apipkg==1.5
aspy.yaml==1.3.0
atomicwrites==1.4.0
attackcti==0.3.4.3
attrs==21.2.0
azure-common==1.1.27
azure-core==1.18.0
azure-mgmt-compute==23.0.0
azure-identity==1.6.1
azure-mgmt-core==1.2.1
azure-mgmt-network==19.0.0
azure-mgmt-resource==17.0.0
bcrypt==3.2.0
boto3==1.18.38
botocore==1.22.5
certifi==2021.5.30
cffi==1.15.0
cfgv==3.3.0
chardet==4.0.0
colorama==0.4.4
configparser==5.0.2
contextlib2==0.6.0.post1
Deprecated==1.2.13
dnspython==2.1.0
docutils==0.18
execnet==1.9.0
gitdb==4.0.5
GitPython==3.1.24
identify==2.2.10
idna==3.3
importlib-metadata==4.0.1
Jinja2==3.0.2
jmespath==0.10.0
lockfile==0.12.2
MarkupSafe==2.0.1
mock==4.0.3
more-itertools==8.9.0
nodeenv==1.6.0
mysql-connector-python==8.0.26
ntlm-auth==1.5.0
packaging==21.0
path==15.1.2
path.py==12.5.0
pexpect==4.8.0
pluggy==0.13.1
pre-commit==2.15.0
protobuf==3.18.1
psutil==5.8.0
ptyprocess==0.7.0
py==1.10.0
pycparser==2.20
PyGithub==1.55
PyInquirer==1.0.3
PyJWT==2.1.0
PyNaCl==1.4.0
pyparsing==2.4.7
pyperclip==1.8.2
pytest==6.2.5
python-daemon==2.3.0
python-dateutil==2.8.1
python-terraform==0.10.1
pywinrm==0.4.2
PyYAML==6.0
requests==2.25.1
requests-ntlm==1.1.0
s3transfer==0.5.0
six==1.16.0
smmap==3.0.5
splunk-sdk==1.6.16
tabulate==0.8.9
termcolor==1.1.0
toml==0.10.2
urllib3<1.26.8
virtualenv==20.9.0
wcwidth==0.2.5
wget==3.2
wrapt==1.13.1
xmltodict==0.12.0
zipp==3.4.1
@@ -1,12 +0,0 @@
This PR was created by Automated Detection Testing Service :robot:
Tested detections:
{% for result in results %}
{{ result.detection_result.detection_name }} {% if result.detection_result.error == 0 %}:white_check_mark:{% else %}:x:{% endif %}
- diskUsage: {{ result.detection_result.diskUsage }}
- runDuration: {{ result.detection_result.runDuration }}s
- scanCount: {{ result.detection_result.scanCount }}
{% endfor %}
@@ -1,3 +0,0 @@
This PR was created by Automated Detection Testing Service :robot:
Please review the dataset.yml and add your data.
-11
View File
@@ -1,11 +0,0 @@
FROM docker.repo.splunkdev.net/ci-cd/ci-container:python-3.7-buster
RUN apt-get update
RUN apt-get install -y python3-dev git python-dev unzip python3-pip
RUN apt-get install -y python-gitdb
WORKDIR /app
RUN pip3 install -r requirements.txt
ENTRYPOINT ["python3", "detection_service.py"]
CMD ["-tfn", "T1003_002"]
-28
View File
@@ -1,28 +0,0 @@
# SSA Testing
The SSA testing job will test SSA detections with the staging tenant research. It will execute the following steps to test SSA detections:
* Create a temporary index for results
* Create a temporary pipeline and decorate it
* Activate a temporary pipeline
* Send raw data to the testing tenant
* Collect the results from the temporary index
* Tear down the test
# SSA Testing GitLab CI
SSA testing job is configured as a GitLAB CI job, which will test all changed and newly created SSA detections in the given branch compared to the develop branch. For example, if you create two new SSA detections with its corrsponding test files, it will test these two SSA detections in the GitLab CI job. It can take up to 30 min until the GitLab CI job will start.
The GitLab CI job is limited to branches with the prefix ssa*, therefore you need to name your branch ssa* in order to make the GitLab CI job work.
# SSA Testing Manual
Additional to the GitLab CI testing, you can do ad-hoc testing during SSA detection development. First you need, to prepare your virtualenv:
````
cd bin/ssa-end-to-end-testing
virtualenv -p python3 venv && source venv/bin/activate && pip3 install -r requirements.txt
````
Then you can run the SSA detection test (please consider that your detections needs to be pushed to the GitHub repository):
````
python run_ssa_smoketest.py -e staging -s research -b [your_branch] -t [your_token]
````
The token can be derived from the SSA tenant.
In order to only test a single detection, you can run the following command:
````
python run_ssa_smoketest.py -e staging -s research -b [your_branch] -tf endpoint/ssa___detect_pass_hash.test.yml -t [your_token]
````
-18
View File
@@ -1,18 +0,0 @@
def pytest_addoption(parser):
parser.addoption("--token", action="store", default="scs token")
parser.addoption("--env", action="store", default="scs environment")
parser.addoption("--tenant", action="store", default="environment's tenant")
def pytest_generate_tests(metafunc):
# This is called for every test. Only get/set command line arguments
# if the argument is specified in the list of test "fixturenames".
option_value = metafunc.config.option.token
if 'token' in metafunc.fixturenames and option_value is not None:
metafunc.parametrize("token", [option_value])
env = metafunc.config.option.env
if 'tenant' in metafunc.fixturenames and env is not None:
metafunc.parametrize("env", [env])
tenant = metafunc.config.option.tenant
if 'tenant' in metafunc.fixturenames and tenant is not None:
metafunc.parametrize("tenant", [tenant])
@@ -1,30 +0,0 @@
FROM docker.repo.splunkdev.net/ci-cd/ci-container:debian-buster
ARG SRCBRANCH=develop
ARG SCSENV=app_play1
ARG KUBECTL_VERSION=v1.10.9
ENV QBEC_VERSION 0.12.1
ENV QBEC_SHA 6fbec2c33ff15c603585a22c100aa839461680cf775a1f90336078bf79c6a42f
ENV SRCBRANCH=${SRCBRANCH}
ENV SCSENV=${SCSENV}
RUN mkdir /deployer && \
apt-get update && \
apt-get install -y bash tar curl && \
# Install kubetctl
curl -L https://dl.k8s.io/release/v${KUBECTL_VERSION}/bin/linux/amd64/kubectl -o /usr/local/bin/kubectl && \
chmod +x /usr/local/bin/kubectl && \
# Install qbec
curl -L -O https://github.com/splunk/qbec/releases/download/v${QBEC_VERSION}/qbec-linux-amd64.tar.gz && \
echo "${QBEC_SHA} qbec-linux-amd64.tar.gz" | sha256sum -c && \
tar -xzf qbec-linux-amd64.tar.gz -C /usr/local/bin qbec && rm -rf qbec-linux-amd64.tar.gz
COPY k8s /deployer/k8s
RUN chmod +x /deployer/k8s/deployer.sh
RUN chmod +x /deployer/k8s/check_deploy_status.sh
WORKDIR /deployer/k8s
@@ -1,60 +0,0 @@
#!/bin/bash
set -o pipefail
set -o nounset
set -o errexit
#This script is used to check the status of the smoketest k8s job: the tests pass if {.status.succeeded} returns 1.
# Then retrieve the job logs once it's completed.
smoketest_k8s_job=$(kubectl get jobs -o json | jq -r --arg JOBID "${CI_JOB_ID}" '.items[] | select(.kind=="Job") | select(.metadata.labels.ciJobId==$JOBID) | .metadata.name')
echo >&2 "Create smoketest k8s job: $smoketest_k8s_job"
smoketest_k8s_pod=$(kubectl get pods -o custom-columns=:metadata.name | grep ${smoketest_k8s_job})
echo >&2 "Create smoketest k8s pod: $smoketest_k8s_pod"
echo >&2 "Checking if smoketest starts running..."
set +o errexit
#Wait for job ready to run as background process, returns 1 if smoketest job still in Pod Initializing stage.
#Will exit if the pod is not up after 5 minutes(max_retries = 30)
job_running=1
counter=1
MAX_RETRIES=30
while [[ $job_running -ne 0 ]] && [[ $counter -le $MAX_RETRIES ]]; do
kubectl wait --for=condition=ready pod/${smoketest_k8s_pod} --timeout=10s
job_running=$?
echo >&2 "Checking if smoketest job starts running (return 0 if the job is ready to run):" $job_running
((counter++))
if [[ $counter -eq $MAX_RETRIES ]]; then
kubectl get pods
echo "Smoke test pod is not up after 5 minutes. Will exit."
exit 1
fi
done
set -o errexit
kubectl logs -f ${smoketest_k8s_pod}
echo >&2 "Complete retrieving smoketest job logs."
#Add more loggings to capture more info for k8s error that happens sporadically
set +o errexit
echo >&2 "Printing command: kubectl get job/${smoketest_k8s_job}"
echo >&2 "$(kubectl get job/${smoketest_k8s_job})"
echo >&2 "$(kubectl get job/${smoketest_k8s_job} -o jsonpath='{.status}')"
echo >&2 "Printing command with extra space: kubectl get job/${smoketest_k8s_job} "
echo >&2 "$(kubectl get job/${smoketest_k8s_job} )"
echo >&2 "$(kubectl get job/${smoketest_k8s_job} -o jsonpath='{.status}')"
echo >&2 "Printing command kubectl get jobs ${smoketest_k8s_job} -o jsonpath='{.status}'"
echo >&2 "$(kubectl get jobs ${smoketest_k8s_job} -o jsonpath='{.status}')"
echo >&2 "*****End of logging for k8s error debugging *****"
set -o errexit
SUCCESS=$(kubectl get job/${smoketest_k8s_job} -o jsonpath='{.status.succeeded}')
if [[ $SUCCESS -ne 1 ]]; then
echo "Smoke test failed. Please refer to the test logs."
exit 1
fi
echo "All smoke tests passed!"
@@ -1,12 +0,0 @@
local params = import '../params.libsonnet';
[
{
apiVersion: 'v1',
kind: 'ServiceAccount',
metadata: {
name: params.components.serviceAccountName,
},
},
]
@@ -1,73 +0,0 @@
local params = import '../params.libsonnet';
local job = {
apiVersion: 'batch/v1',
kind: 'Job',
metadata: {
generateName: 'ssa-smoketest-',
labels: {
name: 'ssa-smoketest',
uploaderLabel: std.extVar('CI_COMMIT_SHORT_SHA'),
ciJobId: std.extVar('CI_JOB_ID'),
},
},
spec: {
activeDeadlineSeconds: 36000,
ttlSecondsAfterFinished: 100,
template: {
metadata: {
annotations: {
'splunk8s.io.vault/init-container': 'true',
},
},
spec: {
serviceAccountName: params.components.serviceAccountName,
restartPolicy: 'Never',
containers: [
{
name: 'ssa-smoke-test',
image: std.extVar('SMOKETEST_RUNNER_IMAGE'),
imagePullPolicy: 'Always',
env: [
{
"name": 'SRCBRANCH',
"value": std.extVar('SRCBRANCH')
},
{
"name": 'SMOKETEST_VAULT_READ_PATH',
"value": params.components.vaultReadPath
},
{
"name": 'DSP_ENV',
"value": params.components.dspEnv
},
{
"name": 'TENANT',
"value": params.components.tenant
},
{
"name": 'ENV',
"value": std.extVar('qbec.io/env')
},
],
resources: {
limits: {
cpu: '1',
memory: '1000Mi'
},
requests: {
cpu: '800m',
memory: '750Mi'
},
},
command: ['/bin/bash', '-c'],
args: ['./run_ssa_smoketest_helper.sh'],
},
],
},
},
},
};
job
@@ -1,3 +0,0 @@
#!/bin/bash
qbec --yes apply $SCSENV --vm:ext-str SRCBRANCH --vm:ext-str CI_COMMIT_SHORT_SHA --vm:ext-str CI_JOB_ID --vm:ext-str SMOKETEST_RUNNER_IMAGE -c service-account -c smoketest --wait
@@ -1,7 +0,0 @@
// this file has the baseline default parameters
{
components: {
serviceAccountName: "sa-tr"
},
}
@@ -1,12 +0,0 @@
// this file has the param overrides for the default environment
local base = import './base.libsonnet';
base {
components +: {
serviceAccountName: "sa-tr-staging",
vaultReadPath: 'scpauth-kube-wicket-iad10/token/threat-research-test.app-gstage1',
tenant: 'research',
dspEnv: 'staging',
}
}
@@ -1,12 +0,0 @@
// this file has the param overrides for the default environment
local base = import './base.libsonnet';
base {
components +: {
serviceAccountName: "sa-tr-playground",
vaultReadPath: 'scpauth-app-play1/token/threat-research-test.app-play1',
tenant: 'research2',
dspEnv: 'playground',
}
}
@@ -1,13 +0,0 @@
// this file returns the params for the current qbec environment
// you need to add an entry here every time you add a new environment.
local env = std.extVar('qbec.io/env');
local paramsMap = {
_: import './environments/base.libsonnet',
app_play1: import './environments/smoke-test-playground.libsonnet',
app_gstaging1: import './environments/smoke-test-gstaging.libsonnet',
};
if std.objectHas(paramsMap, env) then paramsMap[env] else error 'environment ' + env + ' not defined in ' + std.thisFile
@@ -1,20 +0,0 @@
apiVersion: qbec.io/v1alpha1
kind: App
metadata:
name: security-content-stream-smoke-test
spec:
environments:
app_play1:
defaultNamespace: threat-research
server: https://internal.api.app-play1.dev.us-west-2.splunk8s.io
app_gstaging1:
defaultNamespace: tr-st-deploy
server: https://internal.api.kube-wicket-iad10.stage.splunk8s.io
vars:
external:
- name: SCBRANCH
default: develop
secret: false
- name: SMOKE_TEST
default: test
secret: false
@@ -1,5 +0,0 @@
# Add your custom assertion functions for `application` tests in this file
def dummy_application_func(output=list()):
return True
@@ -1,96 +0,0 @@
from sly import Lexer, Parser
import modules.assertions.global_assertions
class AssertionLexer(Lexer):
tokens = {FUNC, NUMBER, STRING, AND, OR, XOR, NOT, LPAR, RPAR, ARG}
ignore = ' \t'
literals = {'&', '|', '^', '!', '(', ')'}
@_(r'\((\s*("[^"]*"|\'[^\']*\'|\d+)\s*,?)+\)')
def ARG(self, t):
arguments = t.value[1:-1].split(",")
args = []
for a in arguments:
a = a.strip()
if a.isnumeric:
args.append(float(a))
else:
args.append(a)
t.value = args
return t
AND = r'\&'
OR = r'\|'
XOR = r'\^'
NOT = r'\!'
LPAR = r'\('
RPAR = r'\)'
FUNC = r'@[a-zA-Z0-9_]+'
STRING = r'("[^"]*"|\'[^\']*\')'
NUMBER = r'\d+'
class AssertionParser(Parser):
tokens = AssertionLexer.tokens
precedence = (
('left', AND, OR, XOR),
('right', NOT)
)
def __init__(self, events):
self.names = {}
self.events = events
@_('FUNC ARG')
def expr(self, p):
f = getattr(modules.assertions.global_assertions, p.FUNC[1:])
return f(output=self.events, *p.ARG)
@_('FUNC LPAR RPAR')
def expr(self, p):
return "%s()" % p.FUNC
@_('expr AND expr')
def expr(self, p):
return p.expr0 and p.expr1
@_('expr OR expr')
def expr(self, p):
return p.expr0 or p.expr1
@_('expr XOR expr')
def expr(self, p):
return p.expr0 != p.expr1
@_('NOT expr')
def expr(self, p):
return not p.expr
@_('LPAR expr RPAR')
def expr(self, p):
return p.expr
@_('NUMBER')
def expr(self, p):
return int(p.NUMBER)
@_('STRING')
def expr(self, p):
return p.STRING
@_('ARG')
def expr(self, p):
return p.ARG
if __name__ == '__main__':
lexer = AssertionLexer()
parser = AssertionParser([1, 2, 3])
code = "@count_gt(0) & (@count_lt(5))"
for tok in lexer.tokenize(code):
print('type=%r, value=%r' % (tok.type, tok.value))
res = parser.parse(lexer.tokenize(code))
print(res)
@@ -1,5 +0,0 @@
# Add your custom assertion functions for `cloud` tests in this file
def dummy_cloud_func(output=[]):
return True
@@ -1,5 +0,0 @@
# Add your custom assertion functions for `endpoint` tests in this file
def dummy_endpoint_func(output=[]):
return True
@@ -1,32 +0,0 @@
# Do not remove these imports that appear unused.
# Functions in this modules are discovered by the assertion_parser through `getattr`
from modules.assertions.application.assertions import *
from modules.assertions.cloud.assertions import *
from modules.assertions.endpoint.assertions import *
from modules.assertions.network.assertions import *
from modules.assertions.web.assertions import *
def count_eq(n, output=[]):
return len(output) == n
def count_lt(n, output=[]):
return len(output) < n
def count_gt(n, output=[]):
return len(output) > n
def count_lte(n, output=[]):
return count_eq(n, output) or count_lt(n, output)
def count_gte(n, output=[]):
return count_eq(n, output) or count_gt(n, output)
def count_not(n, output=[]):
return count_lt(n, output) or count_gt(n, output)
@@ -1,5 +0,0 @@
# Add your custom assertion functions for `network` tests in this file
def dummy_network_func(output=[]):
return True
@@ -1,5 +0,0 @@
# Add your custom assertion functions for `web` tests in this file
def dummy_web_func(output=[]):
return True
@@ -1,41 +0,0 @@
03/12/2021 01:19:11 AM
LogName=Security
SourceName=Microsoft Windows security auditing.
EventCode=4688
EventType=0
Type=Information
ComputerName=win-dc-5236904.attackrange.local
TaskCategory=Process Creation
OpCode=Info
RecordNumber=540110
Keywords=Audit Success
Message=A new process has been created.
Creator Subject:
Security ID: ATTACKRANGE\SuperUser01
Account Name: SuperUser01
Account Domain: ATTACKRANGE
Logon ID: 0x1DB69D
Target Subject:
Security ID: NULL SID
Account Name: -
Account Domain: -
Logon ID: 0x0
Process Information:
New Process ID: 0xd4c
New Process Name: C:\Windows\system32\cmd.exe
Token Elevation Type: %%1936
Mandatory Label: Mandatory Label\High Mandatory Level
Creator Process ID: 0xc88
Creator Process Name: C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe
Process Command Line: "C:\Windows\system32\cmd.exe" /c "reg save HKLM\sam %temp%\sam & reg save HKLM\system %temp%\system & reg save HKLM\Security %temp%\security"
Token Elevation Type indicates the type of token that was assigned to the new process in accordance with User Account Control policy.
Type 3 is a full token with no privileges removed or groups disabled. A full token is only used if User Account Control is disabled or if the user is the built-in Administrator account or a service account.
Type 2 is an elevated token with no privileges removed or groups disabled. An elevated token is used when User Account Control is enabled and the user chooses to start the program using Run as administrator. An elevated token is also used when an application is configured to always require administrative privilege or to always require maximum privilege, and the user is a member of the Administrators group.
Type 3 is a limited token with administrative privileges removed and administrative groups disabled. The limited token is used when User Account Control is enabled, the application does not require administrative privilege, and the user does not choose to start the program using Run as administrator.
File diff suppressed because it is too large Load Diff
@@ -1,123 +0,0 @@
11/24/2021 02:28:14 AM
LogName=Security
SourceName=Microsoft Windows security auditing.
EventCode=4688
EventType=0
Type=Information
ComputerName=win-dc-725.attackrange.local
TaskCategory=Process Creation
OpCode=Info
RecordNumber=257876
Keywords=Audit Success
Message=A new process has been created.
Creator Subject:
Security ID: ATTACKRANGE\Administrator
Account Name: Administrator
Account Domain: ATTACKRANGE
Logon ID: 0x55E621
Target Subject:
Security ID: NULL SID
Account Name: -
Account Domain: -
Logon ID: 0x0
Process Information:
New Process ID: 0xca4
New Process Name: C:\Windows\System32\cmd.exe
Token Elevation Type: %%1936
Mandatory Label: Mandatory Label\High Mandatory Level
Creator Process ID: 0x13a4
Creator Process Name: C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe
Process Command Line: "C:\Windows\system32\cmd.exe" /c "pypykatz live registry"
Token Elevation Type indicates the type of token that was assigned to the new process in accordance with User Account Control policy.
Type 1 is a full token with no privileges removed or groups disabled. A full token is only used if User Account Control is disabled or if the user is the built-in Administrator account or a service account.
Type 2 is an elevated token with no privileges removed or groups disabled. An elevated token is used when User Account Control is enabled and the user chooses to start the program using Run as administrator. An elevated token is also used when an application is configured to always require administrative privilege or to always require maximum privilege, and the user is a member of the Administrators group.
Type 3 is a limited token with administrative privileges removed and administrative groups disabled. The limited token is used when User Account Control is enabled, the application does not require administrative privilege, and the user does not choose to start the program using Run as administrator.
11/24/2021 02:28:15 AM
LogName=Security
SourceName=Microsoft Windows security auditing.
EventCode=4688
EventType=0
Type=Information
ComputerName=win-dc-725.attackrange.local
TaskCategory=Process Creation
OpCode=Info
RecordNumber=257875
Keywords=Audit Success
Message=A new process has been created.
Creator Subject:
Security ID: ATTACKRANGE\Administrator
Account Name: Administrator
Account Domain: ATTACKRANGE
Logon ID: 0x55E621
Target Subject:
Security ID: NULL SID
Account Name: -
Account Domain: -
Logon ID: 0x0
Process Information:
New Process ID: 0x62c
New Process Name: C:\Windows\System32\reg.exe
Token Elevation Type: %%1936
Mandatory Label: Mandatory Label\High Mandatory Level
Creator Process ID: 0xf6c
Creator Process Name: C:\Windows\System32\cmd.exe
Process Command Line: reg save HKLM\security C:\Users\ADMINI~1\AppData\Local\Temp\security
Token Elevation Type indicates the type of token that was assigned to the new process in accordance with User Account Control policy.
Type 1 is a full token with no privileges removed or groups disabled. A full token is only used if User Account Control is disabled or if the user is the built-in Administrator account or a service account.
Type 2 is an elevated token with no privileges removed or groups disabled. An elevated token is used when User Account Control is enabled and the user chooses to start the program using Run as administrator. An elevated token is also used when an application is configured to always require administrative privilege or to always require maximum privilege, and the user is a member of the Administrators group.
Type 3 is a limited token with administrative privileges removed and administrative groups disabled. The limited token is used when User Account Control is enabled, the application does not require administrative privilege, and the user does not choose to start the program using Run as administrator.
11/24/2021 02:28:16 AM
LogName=Security
SourceName=Microsoft Windows security auditing.
EventCode=4688
EventType=0
Type=Information
ComputerName=win-dc-725.attackrange.local
TaskCategory=Process Creation
OpCode=Info
RecordNumber=257874
Keywords=Audit Success
Message=A new process has been created.
Creator Subject:
Security ID: ATTACKRANGE\Administrator
Account Name: Administrator
Account Domain: ATTACKRANGE
Logon ID: 0x55E621
Target Subject:
Security ID: NULL SID
Account Name: -
Account Domain: -
Logon ID: 0x0
Process Information:
New Process ID: 0x1258
New Process Name: C:\Windows\System32\reg.exe
Token Elevation Type: %%1936
Mandatory Label: Mandatory Label\High Mandatory Level
Creator Process ID: 0xf6c
Creator Process Name: C:\Windows\System32\cmd.exe
Process Command Line: reg save HKLM\system C:\Users\ADMINI~1\AppData\Local\Temp\system
Token Elevation Type indicates the type of token that was assigned to the new process in accordance with User Account Control policy.
Type 1 is a full token with no privileges removed or groups disabled. A full token is only used if User Account Control is disabled or if the user is the built-in Administrator account or a service account.
Type 2 is an elevated token with no privileges removed or groups disabled. An elevated token is used when User Account Control is enabled and the user chooses to start the program using Run as administrator. An elevated token is also used when an application is configured to always require administrative privilege or to always require maximum privilege, and the user is a member of the Administrators group.
Type 3 is a limited token with administrative privileges removed and administrative groups disabled. The limited token is used when User Account Control is enabled, the application does not require administrative privilege, and the user does not choose to start the program using Run as administrator.
File diff suppressed because it is too large Load Diff
@@ -1,59 +0,0 @@
import json
from datetime import datetime
from datetime import timedelta
from pytz import timezone
import fileinput
import os
import re
import io
import logging
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
class DataManipulation:
def manipulate_timestamp(self, file_path, sourcetype, source):
if source == 'WinEventLog:Security':
return self.manipulate_timestamp_windows_event_log_raw(file_path)
def manipulate_timestamp_windows_event_log_raw(self, file_path):
path = file_path
f = io.open(path, "r", encoding="utf-8")
pst = timezone('US/Pacific')
self.now = datetime.now(pst)
self.now = self.now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
self.now = datetime.strptime(self.now,"%Y-%m-%dT%H:%M:%S.%fZ")
# read raw logs
regex = r'\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2} [AP]M'
data = f.read()
lst_matches = re.findall(regex, data)
if len(lst_matches) > 0:
latest_event = datetime.strptime(lst_matches[-1],"%m/%d/%Y %I:%M:%S %p")
self.difference = self.now - latest_event
f.close()
result = re.sub(regex, self.replacement_function, data)
with io.open(path + ".swp", "w+", encoding='utf8') as f:
f.write(result)
return path + ".swp"
else:
f.close()
return path
def replacement_function(self, match):
try:
event_time = datetime.strptime(match.group(),"%m/%d/%Y %I:%M:%S %p")
new_time = self.difference + event_time
return new_time.strftime("%m/%d/%Y %I:%M:%S %p")
except Exception as e:
LOGGER.error("Error in timestamp replacement occured: " + str(e))
return match.group()
@@ -1,79 +0,0 @@
import git
import os
import logging
import glob
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
SECURITY_CONTENT_URL = "https://github.com/splunk/security_content"
class GithubService:
def __init__(self, security_content_branch):
self.security_content_branch = security_content_branch
if os.path.exists('security_content'):
LOGGER.warning(f"Found Existing Security Content Project")
self.created_repo = False
else:
self.security_content_repo_obj = self.clone_project(SECURITY_CONTENT_URL, f"security_content", f"develop")
self.security_content_repo_obj.git.checkout(security_content_branch)
self.created_repo = True
def clone_project(self, url, project, branch):
LOGGER.info(f"Clone Security Content Project")
repo_obj = git.Repo.clone_from(url, project, branch=branch)
return repo_obj
def get_changed_test_files_ssa(self):
branch1 = self.security_content_branch
branch2 = 'develop'
g = git.Git('security_content')
changed_ssa_test_files = []
ci_changes = False
if branch1 != 'develop':
differ = g.diff('--name-only', branch1, branch2)
changed_files = differ.splitlines()
for file_path in changed_files:
# added or changed test files
if file_path.startswith('tests'):
if os.path.basename(file_path).startswith('ssa'):
if file_path not in changed_ssa_test_files:
changed_ssa_test_files.append(file_path)
# changed detections
if file_path.startswith('detections'):
if not 'deprecated' in file_path:
if os.path.basename(file_path).startswith('ssa'):
file_path_base = os.path.splitext(file_path)[0].replace('detections', 'tests') + '.test'
file_path_new = file_path_base + '.yml'
if file_path_new not in changed_ssa_test_files:
changed_ssa_test_files.append(file_path_new)
# changed CI code
if file_path == '.gitlab-ci.yml' or file_path.startswith('bin/ssa-end-to-end-testing'):
ci_changes = True
# all SSA test files for nightly build
else:
changed_files = sorted(glob.glob('security_content/tests/*/*.yml'))
for file_path in changed_files:
file_path = file_path.replace('security_content/','')
if os.path.basename(file_path).startswith('ssa'):
changed_ssa_test_files.append(file_path)
# changed CI code
if file_path == '.gitlab-ci.yml' or file_path.startswith('bin/ssa-end-to-end-testing'):
ci_changes = True
return changed_ssa_test_files, ci_changes
@@ -1,58 +0,0 @@
import sys
import yaml
import time
import os
import requests
import shutil
from .data_manipulation import DataManipulation
def load_file(file_path):
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
sys.exit("ERROR: reading {0}".format(file_path))
return file
def prepare_test(file_path):
# read test file and return as object
test_obj = load_file('security_content/' + file_path)
detection_obj = load_file('security_content/detections/' + test_obj['tests'][0]['file'])
# download attack data
epoch_time = str(int(time.time()))
folder_name = "attack_data_" + epoch_time
os.mkdir(folder_name)
for test in test_obj['tests']:
for attack_data in test['attack_data']:
url = attack_data['data']
r = requests.get(url, allow_redirects=True)
attack_data_file_path = folder_name + '/' + attack_data['file_name']
open(folder_name + '/' + attack_data['file_name'], 'wb').write(r.content)
# Update timestamps before replay
if 'update_timestamp' in attack_data:
if attack_data['update_timestamp'] == True:
data_manipulation = DataManipulation()
data_manipulation.manipulate_timestamp(folder_name + '/' + attack_data['file_name'], attack_data['sourcetype'], attack_data['source'])
dict_test = {
"test_obj": test_obj,
"detection_obj": detection_obj,
"attack_data_file_path": attack_data_file_path
}
return dict_test, folder_name
def remove_security_content():
shutil.rmtree('security_content')
def remove_attack_data(file_path):
# remove attack data
shutil.rmtree(file_path)
@@ -1,21 +0,0 @@
| from read_ssa_enriched_events()
| eval timestamp=parse_long(ucast(map_get(input_event, "_time"), "string", null))
| eval process_name=lower(ucast(map_get(input_event, "process_name"), "string", null)),
cmd_line=ucast(map_get(input_event, "process"), "string", null),
dest_user_id=ucast(map_get(input_event, "dest_user_id"), "string", null),
dest_device_id=ucast(map_get(input_event, "dest_device_id"), "string", null)
| where process_name="cmd.exe" OR process_name="reg.exe"
| where cmd_line != null AND
match_regex(cmd_line, /(?i)save\s+/)=true AND
( match_regex(cmd_line, /(?i)HKLM\\Security/)=true OR
match_regex(cmd_line, /(?i)HKLM\\SAM/)=true OR
match_regex(cmd_line, /(?i)HKLM\\System/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\Security/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\SAM/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\System/)=true
)
| eval start_time = timestamp,
end_time = timestamp,
entities = mvappend(dest_device_id, dest_user_id),
body = "TBD"
| into write_ssa_detected_events();
@@ -1,21 +0,0 @@
| from read_ssa_enriched_events()
| eval timestamp=parse_long(ucast(map_get(input_event, "_time"), "string", null))
| eval process_name=lower(ucast(map_get(input_event, "process_name"), "string", null)),
cmd_line=ucast(map_get(input_event, "process"), "string", null),
dest_user_id=ucast(map_get(input_event, "dest_user_id"), "string", null),
dest_device_id=ucast(map_get(input_event, "dest_device_id"), "string", null)
| where process_name="cmd.exe" OR process_name="reg.exe"
| where cmd_line != null AND
match_regex(cmd_line, /(?i)save\s+/)=true AND
( match_regex(cmd_line, /(?i)HKLM\\Security/)=true OR
match_regex(cmd_line, /(?i)HKLM\\SAM/)=true OR
match_regex(cmd_line, /(?i)HKLM\\System/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\Security/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\SAM/)=true OR
match_regex(cmd_line, /(?i)HKEY_LOCAL_MACHINE\\System/)=true
)
| eval start_time = timestamp,
end_time = timestamp,
entities = mvappend(dest_device_id, dest_user_id),
body = "TBD"
| into write_ssa_detected_events();
@@ -1 +0,0 @@
| from read_splunk_firehose();
@@ -1 +0,0 @@
| from read_splunk_firehose();
@@ -1 +0,0 @@
| from read_ssa_enriched_events() | into write_ssa_detected_events();
@@ -1,55 +0,0 @@
import os
import json
import hashlib
#import urllib.request
from requests import get
import re
import logging
from modules.testing_utils import log
SSML_CWD = ".humvee"
HUMVEE_ARTIFACT_SEARCH = "https://repo.splunk.com/artifactory/api/search/artifact?name=humvee&repos=maven-splunk-local"
def get_latest_humvee_object():
#res = json.loads(urllib.request.urlopen(HUMVEE_ARTIFACT_SEARCH).read().decode('utf-8'))
res = get(HUMVEE_ARTIFACT_SEARCH).json()
for r in res['results']:
if re.match(r".*/latest/humvee-.*\.jar$", r['uri']):
#latest_humvee = json.loads(urllib.request.urlopen(r['uri']).read().decode('utf-8'))
latest_humvee = get(r['uri']).json()
return latest_humvee
return ""
def build_humvee(path):
if not os.path.exists(path):
os.mkdir(path)
latest_humvee_object = get_latest_humvee_object()
humvee_path = "%s/humvee.jar" % path
humvee_sha256 = ""
if os.path.exists(humvee_path):
with open(humvee_path, 'rb') as jar_fh:
humvee_sha256 = hashlib.sha256(jar_fh.read()).hexdigest()
log(logging.DEBUG, "Current local checksum of Humvee", detail=humvee_sha256)
if humvee_sha256 != latest_humvee_object['checksums']['sha256']:
log(logging.INFO, "Downloading Latest Humvee")
log(logging.DEBUG, "Humvee details", detail=latest_humvee_object)
#urllib.request.urlretrieve(latest_humvee_object['downloadUri'], humvee_path)
with open(humvee_path, 'wb') as f:
f.write(get(latest_humvee_object['downloadUri']).content)
else:
log(logging.DEBUG, "Already latest checksum %s" % humvee_sha256, detail=latest_humvee_object)
#def convert_to_ssa(detection):
# '''
# curl -H 'Content-type: text/yaml' -H 'Authorization: Bearer TOKEN'
# https://app-admin.playground.scp.splunk.com/secanalytics/ssa-tenant-management/v1alpha1/admin/detection-spl/research2
# --data-binary @detections/endpoint/ssa___first_time_seen_cmd_line.yml
# @param detection:
# @return:
# '''
@@ -1,580 +0,0 @@
"""
helper functions to use the Streams Service API (v3beta1) to perform create, read and delete operations on
data pipeline.
API doc: https://dev.splunk.com/enterprise/reference/api/streams/v3beta1
"""
import logging
import os
import uuid
import requests
import time
import base64
import json
from http import HTTPStatus
#from constants import ML_MODEL_CONNECTOR_UUID
from modules.utils import request_headers
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
TENANT_PLAYGROUND = f"research2"
TENANT_STAGING = f"research"
BASE_URL_PLAYGROUND = f"https://api.playground.scp.splunk.com/"
BASE_URL_STAGING = f"https://api.staging.scs.splunk.com/"
# Streaming Pipelines REST endpoints
CONNECTIONS_ENDPOINT = f"streams/v3beta1/connections"
PIPELINES_ENDPOINT = f"streams/v3beta1/pipelines"
PIPELINES_COMPILE_ENDPOINT = f"streams/v3beta1/pipelines/compile"
PIPELINES_VALIDATE_ENDPOINT = f"streams/v3beta1/pipelines/validate"
PIPELINES_REGISTRY_ENDPOINT = f"streams/v3beta1/pipelines/registry"
PREVIEW_SESSION_ENDPOINT = f"streams/v3beta1/preview-session"
PREVIEW_DATA_ENDPOINT = f"streams/v3beta1/preview-data"
INGEST_ENDPOINT = f"ingest/v1beta2/events"
SUBMIT_SEARCH_ENDPOINT = f"search/v2beta1/jobs"
DATASETS_ENDPOINT = f"catalog/v2beta1/datasets"
class ApiError(Exception):
pass
class DSPApi:
def __init__(self, env, tenant, token):
self.env = env
self.tenant = tenant
self.header_token = f"Bearer {token}"
self.validate_token()
def validate_token(self):
def decode(s):
def pad(t):
return t + '=' * (len(t) % 4)
return json.loads(base64.b64decode(pad(s)))
token = self.header_token.split()[1]
header, payload, signature = token.split('.')
payload_data = decode(payload)
for k in sorted(payload_data.keys()):
LOGGER.info(f"token.payload.{k} = %s", payload_data[k])
valid_for = payload_data['exp'] - int(time.time())
if not valid_for > 0:
raise ApiError("Token is expired")
token_env = payload_data['iss'].split('.')[-4]
if self.env != token_env:
raise ApiError(f"Env {self.env} was specified but token is for {token_env}")
token_tenant = payload_data['tenant']
if self.tenant != token_tenant:
raise ApiError(f"Tenant {self.tenant} was specified but token is for {token_tenant}")
def return_api_endpoint(self, endpoint):
if self.env == 'playground':
return f"{BASE_URL_PLAYGROUND}{TENANT_PLAYGROUND}/{endpoint}"
else:
return f"{BASE_URL_STAGING}{TENANT_STAGING}/{endpoint}"
def compile_spl(self, spl):
"""
Compile SPL text to a UPL JSON
Parameters
----------
spl: str
the SPL representation of a pipeline or function parameter to be compiled
Returns
-------
upl:
JSON representation of the compiled AST
"""
data = {"spl": spl}
LOGGER.debug(f"Compiling SPL into UPL")
#LOGGER.info(f"{spl}")
response = requests.post(self.return_api_endpoint(PIPELINES_COMPILE_ENDPOINT), json=data, headers=request_headers(self.header_token))
upl = response.json()
if response.status_code == HTTPStatus.OK:
LOGGER.info(f"Successfully compiled spl to upl")
return upl
else:
LOGGER.error("SPL compilation failed: %s", response.text)
def validate_upl(self, upl):
"""
Validate whether the JSON representation of a pipeline is valid
Parameters
----------
upl: JSON
JSON representation of the compiled AST of a pipeline
Returns
-------
response_body: JSON
returns whether or not the pipeline id valid. If valid, the response body returns 'success'
"""
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
data = {"upl": upl}
response = requests.post(self.return_api_endpoint(PIPELINES_VALIDATE_ENDPOINT), json=data, headers=headers)
if response.status_code == HTTPStatus.OK:
LOGGER.info(f"UPL is validated.")
return upl
else:
LOGGER.error("UPL validation failed: %s", response.text)
def get_pipelines(self):
"""
Returns the list of pipelines
@return:
list of pipelines
"""
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
response = requests.get(self.return_api_endpoint(PIPELINES_ENDPOINT), headers=headers)
response_body = response.json()
if response.status_code == HTTPStatus.OK:
return response_body.get('items')
else:
LOGGER.error(f"Failed to get pipelines: %s", response.text)
def create_pipeline(self, upl):
"""
POST pipelines endpoint to create a pipeline based on the valid upl
Parameters
----------
upl: JSON
JSON representation of the validated pipeline details
Returns
-------
pipline_id: UUID
id of the created pipeline
"""
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
set_test_id = uuid.uuid4().hex
data = {
"name": f"ssa_smoke_test_pipeline_helper_{set_test_id}",
"description": "ssa_test_pipeline_description",
"bypassValidation": "true",
"labels": {"app": "ba"},
"data": upl
}
response = requests.post(self.return_api_endpoint(PIPELINES_ENDPOINT), json=data, headers=headers)
response_body = response.json()
if response.status_code == HTTPStatus.CREATED:
pipeline_id = response_body.get("id")
#LOGGER.info(f"Pipeline {pipeline_id} successfully created")
return pipeline_id
else:
LOGGER.error(f"Failed to create pipeline: %s", response.text)
def create_pipeline_from_spl(self, spl):
"""
helper function to compile and validate from spl text, then create the pipeline
"""
upl = self.compile_spl(spl)
validated_upl = self.validate_upl(upl)
pipeline_id = self.create_pipeline(validated_upl)
LOGGER.info(f"pipeline id created is: {pipeline_id}")
return pipeline_id
def activate_pipeline(self, pipeline_id):
"""
POST pipelines/activate endpoint to activate an existing pipeline
Parameters
----------
pipeline_id: str
pipeline UUID to activate
Returns
-------
response_body: response JSON
response body that contains pipeline status ACTIVATED
"""
assert(pipeline_id is not None), "Must specify a 'pipeline_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
pipelines_activate_endpoint = self.return_api_endpoint(PIPELINES_ENDPOINT) + "/" + pipeline_id + "/activate"
pipelines_status_endpoint = self.return_api_endpoint(PIPELINES_ENDPOINT) + "/" + pipeline_id
data = {
"activateLatestVersion": "true",
"allowNonRestoredState": "true",
"skipRestoreState": "true"
}
pipeline_activated = False
attempts_remaining = 30
response = requests.post(pipelines_activate_endpoint, json=data, headers=headers)
if response.status_code == HTTPStatus.OK:
while attempts_remaining:
attempts_remaining -= 1
pipeline_status_response = requests.get(pipelines_status_endpoint, headers=headers)
if pipeline_status_response.status_code == HTTPStatus.OK:
pipeline_status = pipeline_status_response.json()
status = pipeline_status['status']
if status == 'ACTIVATED':
pipeline_activated = True
LOGGER.info(f"Pipeline {pipeline_id} successfully activated")
break
else:
LOGGER.warning("Current pipeline activation status for %s: %s", pipeline_id, status)
else:
LOGGER.error("Failed to check pipeline status for %s: %s", pipeline_id, pipeline_status_response.text)
if attempts_remaining:
time.sleep(60)
else:
LOGGER.error("Got tired of waiting for the pipeline to activate")
else:
LOGGER.error("Failed to request pipeline activation for %: %s", pipeline_id, response.text)
return pipeline_activated
def deactivate_pipeline(self, pipeline_id):
"""
POST pipelines/deactivate endpoint to deactivate an existing pipeline
Parameters
----------
pipeline_id:
pipeline UUID to deactivate
Returns
-------
response_body: response JSON
response body that contains the status of the pipeline after deactivation
"""
assert(pipeline_id is not None), "Must specify a 'pipeline_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
pipelines_deactivate_endpoint = self.return_api_endpoint(PIPELINES_ENDPOINT) + "/" + pipeline_id + "/deactivate"
data = {
"skipSavepoint": "true"
}
response = requests.post(pipelines_deactivate_endpoint, json=data, headers=headers)
response_body = response.json()
return response, response_body
def delete_pipeline(self, pipeline_id):
"""
Delete an existing pipeline using its pipeline UUID
Parameters
----------
pipeline_id:
pipeline UUID of an existing pipeline
Returns
-------
response status code
"""
assert(pipeline_id is not None), "Must specify a 'pipeline_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
delete_pipeline_endpoint = self.return_api_endpoint(PIPELINES_ENDPOINT) + "/" + pipeline_id
response = requests.delete(delete_pipeline_endpoint, headers=headers)
LOGGER.info(f"DELETE pipeline response status code is: {response.status_code}")
return response
def pipeline_status(self, pipeline_id):
"""
Returns the statues of an existing pipeline
Parameters
----------
pipeline_id:
pipeline UUID of an existing pipeline.
Returns
-------
pipeline_status: str
pipeline status can be CREATED, ACTIVATED, FINISHED, RESTARTING, FAILED.
"""
assert(pipeline_id is not None), "Must specify a 'pipeline_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
pipelines_status_endpoint = self.return_api_endpoint(PIPELINES_ENDPOINT) + "/" + pipeline_id
response = requests.get(pipelines_status_endpoint, headers=headers)
response_body = response.json()
if response.status_code == HTTPStatus.OK:
pipeline_status = response_body.get("status")
return pipeline_status
else:
LOGGER.error(f"Fail to get current status of pipeline pipeline {pipeline_id}")
return
def get_preview_id(self, upl):
"""
POST preview-session endpoint to create a preview session for a pipeline
Parameters
----------
upl: JSON
JSON representation of the pipeline details
Returns
-------
preview_id: str
preview session id
"""
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
data = {
"upl": upl
}
response = requests.post(self.return_api_endpoint(PREVIEW_SESSION_ENDPOINT), json=data, headers=headers)
response_body = response.json()
# The preview sessions was started successfully
if response.status_code == HTTPStatus.CREATED:
preview_id = response_body.get("previewId")
assert(preview_id is not None), "Must return a 'preview_id'"
return preview_id
def get_preview_data(self, preview_id):
"""
POST preview-data endpoint to get the preview data for a preview session
Parameters
----------
preview_id: str
preview session id
Returns
-------
response_body: response JSON
the response would contain current number of records, preview data
"""
assert(preview_id is not None), "Must specify a 'preview_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
preview_data_endpoint = self.return_api_endpoint(PREVIEW_DATA_ENDPOINT) + "/" + str(preview_id)
response = requests.get(preview_data_endpoint, headers=headers)
response_body = response.json()
if response.status_code != HTTPStatus.OK:
LOGGER.error(f"Failed to preview data from the pipeline. Please check if the operator has been properly "
f"uploaded to DSP.")
return
else:
return response, response_body
def stop_preview_session(self, preview_id):
"""
Delete an existing pipeline using its pipeline UUID
Parameters
----------
preview_id: str
preview session id
Returns
-------
Boolean: True if preview session is stopped.
"""
assert(preview_id is not None), "Must specify a 'preview_id'"
headers = {"Content-Type": "application/json", "Authorization": self.header_token}
stop_preview_session_endpoint = self.return_api_endpoint(PREVIEW_SESSION_ENDPOINT) + "/" + str(preview_id)
response = requests.delete(stop_preview_session_endpoint, headers=headers)
LOGGER.info(f"DELETE/preview-session response status code is: {response.status_code}")
return response
def get_preview_id_from_spl(self, spl):
"""
helper function to compile and validate from spl text, then create the pipeline preview session
"""
upl = self.compile_spl(spl)
validated_upl = self.validate_upl(upl)
preview_id = self.get_preview_id(validated_upl)
LOGGER.info(f"preview id created is: {preview_id}")
return preview_id
def ingest_data(self, data, sourcetype):
"""
Send events
Parameters
----------
data: str
datafile
Returns
-------
response
response body in JSON format
"""
if sourcetype == "WinEventLog:Security":
sourcetype = "WinEventLog"
data = [{
"body": event,
"sourcetype": sourcetype
} for event in data]
response = requests.post(self.return_api_endpoint(INGEST_ENDPOINT), json=data, headers=request_headers(self.header_token))
if response.status_code != HTTPStatus.OK:
LOGGER.error(f"Failed to upload data: %s", response.text)
return False
return True
def submit_search_job(self, module, query):
"""
Submit Search job
Parameters
----------
module: str
module where this query will be run on (default: `mc`)
query: str
SPL of the search query
Returns
-------
sid
sid to search job
"""
data = {
"query": query,
"module": module
}
LOGGER.info(f"Submit Search Job")
response = requests.post(self.return_api_endpoint(SUBMIT_SEARCH_ENDPOINT), json=data, headers=request_headers(self.header_token))
if response.status_code != HTTPStatus.CREATED:
LOGGER.error(f"Submit search job failed: %s", response.text)
return None
else:
response_body = response.json()
return response_body.get("sid")
def check_search_job_finished(self, sid):
"""
Check if a search job finished
Parameters
----------
sid: str
ID of the search. Returned value from `submit_search_job`
Returns
-------
response
boolean true or false
"""
LOGGER.info(f"Check Search job results")
results_check_search_job = self.return_api_endpoint(SUBMIT_SEARCH_ENDPOINT) + "/" + sid
response = requests.get(results_check_search_job, headers=request_headers(self.header_token))
if response.status_code != HTTPStatus.OK:
LOGGER.error(f"Failed to get status of search job")
return None
else:
response_json = response.json()
LOGGER.info(f"Check if search is finished.")
if response_json.get("status") == "done":
return True
else:
return False
def get_search_job_results(self, sid):
"""
Get Search Job Results
Parameters
----------
sid: str
ID of the search. Returned value from `submit_search_job`
Returns
-------
results
results of search
"""
LOGGER.info(f"Get Search job results")
results_search_job_endpoint = self.return_api_endpoint(SUBMIT_SEARCH_ENDPOINT) + "/" + sid + "/results"
response = requests.get(results_search_job_endpoint, headers=request_headers(self.header_token))
if response.status_code != HTTPStatus.OK:
LOGGER.error(f"Failed to get search results")
return None
else:
response_body = response.json()
return response_body.get("results")
def create_temp_index(self, module):
"""
Creates an index under module
Parameters
@param module: str
module under this index will be created
@return:
index object dictionary
"""
index_name = f"temp_st_{uuid.uuid1()}".replace("-", "_")
data = {
"module": module,
"name": index_name,
"kind": "index",
"disabled": False
}
response = requests.post(self.return_api_endpoint(DATASETS_ENDPOINT), headers=request_headers(self.header_token), json=data)
LOGGER.info(f"Create Temp Index {index_name}")
return response.json()
def delete_temp_index(self, index_id):
"""
Deletes an index
@param index_id:
Index ID
@return:
response status code from API
"""
LOGGER.info(f"Delete Temp Index")
datasets_endpoint_api = self.return_api_endpoint(DATASETS_ENDPOINT)
delete_url = f"{datasets_endpoint_api}/{index_id}"
response = requests.delete(delete_url, headers=request_headers(self.header_token))
return response.status_code
@@ -1,258 +0,0 @@
import logging
import os
from re import S
import time
import sys
import uuid
from http import HTTPStatus
from modules.streams_service_api_helper import DSPApi
from modules.utils import check_source_sink, manipulate_spl, read_spl, read_data
from ssa_test import assert_results
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
# MACROS
SLEEP_TIME_CREATE_INDEX = 10
SLEEP_TIME_ACTIVATE_PIPELINE = 10
SLEEP_TIME_SEND_DATA = 30
WAIT_CYCLE = 20
MAX_EXECUTION_TIME_LIMIT = 600 # per detection test
TEST_DATASET = 'windows-security_debug.txt'
class SSADetectionTesting:
def __init__(self, env, tenant, token):
self.execution_passed = True
self.max_execution_time = MAX_EXECUTION_TIME_LIMIT
self.env = env
self.tenant = tenant
self.api = DSPApi(env, tenant, token)
self.test_results = {}
def test_dsp_pipeline(self):
file_path_data = os.path.join(os.path.dirname(__file__), 'data', TEST_DATASET)
file_path_spl = os.path.join(os.path.dirname(__file__), 'spl')
test_spls = ['troubleshoot.spl', 'detection.spl']
test_names = [
"SSA Index Test Minimal",
"SSA Index Detection Testing Example"
]
test_results = []
for i in range(0, len(test_spls)):
self.max_execution_time = MAX_EXECUTION_TIME_LIMIT
test_id = str(uuid.uuid4())
test_result = self.ssa_detection_test(read_spl(file_path_spl, test_spls[i]), file_path_data, test_names[i], test_id, "WinEventLog")
test_results.append(test_result.copy())
passed = True
LOGGER.info('-----------------------------------')
LOGGER.info('-------- test DSP Pipeline --------')
LOGGER.info('-----------------------------------')
for test_result in test_results:
LOGGER.info(test_result['msg'])
passed = passed and test_result['result']
LOGGER.info('-----------------------------------')
return passed
def test_ssa_detections(self, test_obj):
LOGGER.info('Test SSA Detection: ' + test_obj["detection_obj"]["name"])
self.max_execution_time = MAX_EXECUTION_TIME_LIMIT
file_path_attack_data = test_obj["attack_data_file_path"]
test_id = str(uuid.uuid4())
test_results = self.ssa_detection_test(test_obj["detection_obj"]["search"], file_path_attack_data,
"SSA Smoke Test " + test_obj["test_obj"]["name"], test_id,
test_obj['test_obj']['tests'][0]['attack_data'][0]['source'],
test_obj['test_obj']['tests'][0]['pass_condition'])
return test_results
## Helper Functions ##
def update_execution_time(self, time_frame):
self.max_execution_time = self.max_execution_time - WAIT_CYCLE
if self.max_execution_time < 0:
return True
else:
return False
def wait_time(self, time_in_s):
time.sleep(time_in_s)
return self.update_execution_time(time_in_s)
def ssa_detection_test_init(self):
self.cleanup_old_pipelines()
self.test_results["result"] = True
self.test_results["msg"] = ""
#self.results_index = self.api.create_temp_index("mc")
self.created_pipelines = []
self.activated_pipelines = []
def cleanup_old_pipelines(self):
pipelines = self.api.get_pipelines()
yesterday = (time.time() - 24*3600) * 1000 # milliseconds
for pipeline in pipelines:
if pipeline['name'].startswith("ssa_smoke_test_pipeline_helper") and pipeline['createDate'] < yesterday:
if pipeline['status'] == 'ACTIVATED':
# deactivate pipeline
resp, _ = self.api.deactivate_pipeline(pipeline['id'])
if resp.status_code != HTTPStatus.OK:
LOGGER.error("Error deactivating old pipeline %s: %s", pipeline['name'], resp.text)
# delete pipeline
resp = self.api.delete_pipeline(pipeline['id'])
if resp.status_code != HTTPStatus.NO_CONTENT:
LOGGER.error("Error deleting old pipeline %s: %s", pipeline['name'], resp.text)
else:
LOGGER.warning("Found and deleted an old pipeline: %s", pipeline['name'])
def ssa_detection_test_main(self, spl, source, test_name, pass_condition, test_id, sourcetype):
self.execution_passed = True
self.wait_time(SLEEP_TIME_CREATE_INDEX)
check_ssa_spl = check_source_sink(spl)
spl = manipulate_spl(self.api.env, spl, test_id)
assert spl is not None, "fail to manipulate spl file"
upl = self.api.compile_spl(spl)
assert upl is not None, "failed to compile spl"
validated_upl = self.api.validate_upl(upl)
assert validated_upl is not None, "failed to validate upl"
pipeline_id = self.api.create_pipeline(validated_upl)
assert pipeline_id is not None, "failed to create a pipeline"
_pipeline_status = self.api.pipeline_status(pipeline_id)
assert _pipeline_status == "CREATED", f"Current status of pipeline {pipeline_id} should be CREATED"
self.created_pipelines.append(pipeline_id)
pipeline_activated = self.api.activate_pipeline(pipeline_id)
assert pipeline_activated, f"pipeline {pipeline_id} should be activated."
self.activated_pipelines.append(pipeline_id)
self.wait_time(SLEEP_TIME_ACTIVATE_PIPELINE)
if not check_ssa_spl:
msg = f"Detection test successful for {test_name}"
LOGGER.warning(f"Test not completed. Detection seems deprecated, and will not send messages to SSA")
self.test_results["msg"] = msg
return self.test_results
data = read_data(source, sourcetype)
LOGGER.info("Sending (%d) events" % (len(data)))
assert len(data) > 0, "No events to send, skip to next test."
data_uploaded = self.api.ingest_data(data, sourcetype)
assert data_uploaded, "Failed to upload test data"
self.wait_time(SLEEP_TIME_SEND_DATA)
search_results = False
max_execution_time_reached = False
while not (search_results or max_execution_time_reached):
max_execution_time_reached = self.wait_time(WAIT_CYCLE)
query = f"from indexes('detection_testing') | search test_id=\"{test_id}\" "
LOGGER.info(f"Executing search query: {query}")
sid = self.api.submit_search_job('mc', query)
assert sid is not None, f"Failed to create a Search Job"
job_finished = False
while not job_finished:
self.wait_time(WAIT_CYCLE)
result = self.api.check_search_job_finished(sid)
job_finished = result
results = self.api.get_search_job_results(sid)
search_results = (len(results) > 0)
if not search_results:
LOGGER.info(
f"Search didn't return any results. Retrying in {WAIT_CYCLE}s, max execution time left {self.max_execution_time}s")
if not results:
LOGGER.warning("Search job didn't return any results")
LOGGER.info('Received %s result(s)', len(results))
test_passed = assert_results(pass_condition, results)
assert test_passed, f"Pass condition {pass_condition} not satisfied"
msg = f"Detection test successful for {test_name}"
LOGGER.info(msg)
self.test_results["msg"] = msg
return self.test_results
def ssa_detection_test_teardown(self):
"""
Deactivate and deletes pipelines, deletes results indexes,
and when it fails it shows pipelines and result indexes that were not removed.
:return:
None
"""
deactivate_pipeline = lambda p: self.api.deactivate_pipeline(p)[0].status_code == HTTPStatus.OK
delete_pipeline = lambda p: self.api.delete_pipeline(p).status_code == HTTPStatus.NO_CONTENT
#delete_index = lambda p: self.api.delete_temp_index(p["id"]) == HTTPStatus.NO_CONTENT
self.activated_pipelines = [p for p in self.activated_pipelines if not deactivate_pipeline(p)]
self.created_pipelines = [p for p in self.created_pipelines if not delete_pipeline(p)]
if len(self.activated_pipelines) > 0 or len(self.created_pipelines) > 0:
LOGGER.warning("Not all SCS resources freed up")
LOGGER.info(f"Created Pipelines: {','.join(self.created_pipelines)}")
LOGGER.info(f"Active Pipelines: {','.join(self.activated_pipelines)}")
LOGGER.info(f"Result Indexes: {self.results_index}")
else:
LOGGER.info("Testing successfully cleaned up")
def ssa_detection_test(self, spl, source, test_name, test_id, sourcetype, pass_condition='@count_gt(0)'):
self.ssa_detection_test_init()
try:
test_result = self.ssa_detection_test_main(spl, source, test_name, pass_condition, test_id, sourcetype)
self.ssa_detection_test_teardown()
return test_result
except AssertionError as e:
self.ssa_detection_test_teardown()
LOGGER.error(e.args[0])
LOGGER.error(f"Detection test failure for {test_name}")
return {"result": False,
"msg": f"Detection test failure for {test_name}"}
except Exception as e:
self.ssa_detection_test_teardown()
LOGGER.exception(f"Detection test failure for {test_name} (perhaps SCS problems)")
return {"result": False,
"msg": f"Detection test failure for {test_name} (perhaps SCS problems)"}
# only for troubleshooting
# def ssa_detection_in_dsp_with_preview_session(self, spl, source, test_name):
# self.execution_passed = True
# spl = manipulate_spl(self.api.env, spl)
# self.check_result(spl is not None, "fail to read dummy spl file")
# preview_id = self.api.get_preview_id_from_spl(spl)
# self.check_result(preview_id is not None, "failed to create a preview session %s" % spl)
# time.sleep(SLEEP_TIME_SHORT)
# data = read_data(source)
# response_body = self.api.ingest_data(data)
# time.sleep(SLEEP_TIME_LONG)
# response, response_body = self.api.get_preview_data(preview_id)
# self.check_result(response_body.get("currentNumberOfRecords") > 0, "Missing records in preview session.")
# response = self.api.stop_preview_session(preview_id)
# self.write_test_results(test_name)
@@ -1,41 +0,0 @@
import os
import sys
import logging
import coloredlogs
#import urllib.request
from requests import get
import yaml
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(coloredlogs.ColoredFormatter("%(asctime)s - %(levelname)s - %(message)s%(detail)s"))
logger.addHandler(handler)
def get_path(p):
return os.path.join(os.path.join(os.path.dirname(__file__), "..", p))
def log(level, msg, detail=None):
args = {'detail': ""} if detail is None else {'detail': "\n%s" % detail}
logger.log(level, msg, extra=args)
def get_detection(unit_test):
with open(get_path("../detections/%s" % unit_test['file'])) as detection_fh:
parsed_detection = yaml.safe_load(detection_fh)
return parsed_detection
def pull_data(test, destination):
data_desc = dict()
if 'attack_data' in test:
for d in test['attack_data']:
test_data = "%s/%s" % (destination, d['file_name'])
#urllib.request.urlretrieve(d['data'], test_data)
with open(test_data, 'wb') as f:
f.write(get(d['data']).content)
data_desc[d['file_name']] = test_data
log(logging.DEBUG, "Downloading dataset %s from %s" % (d['file_name'], d['data']))
return data_desc
-145
View File
@@ -1,145 +0,0 @@
import json
import logging
import os
import fileinput
import re
import io
from .data_manipulation import DataManipulation
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
# Macros
PULSAR_SOURCE_CONNECTION_ID_PLAYGROUND = f"29fb61f1-9342-48f5-9793-1afa008c377b"
PULSAR_SOURCE_TOPIC_PLAYGROUND = f"persistent://ssa/egress/decorated-events-research2"
PULSAR_SOURCE_CONNECTION_ID_STAGING = f"d156ed3a-1254-469a-bffc-9dd6eda97ac6"
PULSAR_SOURCE_TOPIC_STAGING = f"persistent://ssa/egress/decorated-events-research"
READ_SSA_ENRICHED_EVENTS_EXPANDED = (
f"pulsar(\"__PULSAR_SOURCE_CONNECTION_ID__\", \"__PULSAR_SOURCE_TOPIC__\")"
f"| eval input_event=deserialize_json_object(value)"
f"| select input_event"
f"| eval _datamodels=ucast(map_get(input_event, \"_datamodels\"), \"collection<string>\", [])"
f",body={{}}"
)
# not used in the moment
# PULSAR_SINK_CONNECTION_ID = f"29fb61f1-9342-48f5-9793-1afa008c377b"
# PULSAR_SINK_TOPIC = f"persistent://ssa/ingress/detection-events-research2"
# ## dummy values ##
# DETECTION_TYPE = f"anomaly"
# DETECTION_ID = f"93fbec4e-0375-440c-8db3-4508eca470c4"
# DETECTION_VERSION = f"1"
# RISK_SEVERITY = f"low"
# ## dummy values ##
# WRITE_SSA_DETECTED_EVENTS_EXPANDED = (
# f"| eval create_time=time()"
# f", type=\"{DETECTION_TYPE}\""
# f", detection_id=\"{DETECTION_ID}\""
# f", detection_version=\"{DETECTION_VERSION}\""
# f", risk_severity=\"{RISK_SEVERITY}\""
# f"| select create_time"
# f", start_time"
# f", end_time"
# f", type"
# f", detection_id"
# f", detection_version"
# f", risk_severity"
# f", entities"
# f", body"
# f"| eval id=concat(\"sha256:\", base64_encode(sha256(serialize_json())))"
# f"| select id, serialize_json() AS value"
# f"| into pulsar(\"{PULSAR_SINK_CONNECTION_ID}\", \"{PULSAR_SINK_TOPIC}\", id, value);"
# )
def fetch_token(file):
with open(file) as f:
data = json.load(f)
try:
token = data["data"]["token"]
except RuntimeError as e:
LOGGER.error(f"JSON is invalid, no IAC token found in Vault JSON output: {e}")
return token
def request_headers(header_token):
headers = {"Content-Type": "application/json", "Authorization": header_token}
return headers
def check_source_sink(spl):
match_sink = re.search(r"\|\s*into\s+write_ssa_detected_events\(\s*\)\s*;", spl)
return match_sink
def manipulate_spl(env, spl, test_id):
# Obtain the SSA source
pulsar_source_connection_id, pulsar_source_topic = return_macros(env)
source = READ_SSA_ENRICHED_EVENTS_EXPANDED\
.replace("__PULSAR_SOURCE_CONNECTION_ID__", pulsar_source_connection_id)\
.replace("__PULSAR_SOURCE_TOPIC__", pulsar_source_topic)
# Obtain the test sink
sink = f" eval test_id=\"{test_id}\" | into index(\"mc\", \"detection_testing\")"
# Replace spl template with its `source` and `sink`
spl = replace_ssa_macros(source, sink, spl)
LOGGER.info(f"spl: {spl}")
return spl
def read_spl(file_path, file_name):
full_path = os.path.join(file_path, file_name)
spl = open(full_path, "r").read()
return spl
def replace_ssa_macros(source, sink, spl):
spl = re.sub(r'read_ssa_enriched_events\(\s*\)', source, spl, flags=re.IGNORECASE)
spl = re.sub(r'into write_ssa_detected_events()\(\s*\)', sink, spl, flags=re.IGNORECASE)
return spl
def read_data(file_path, sourcetype):
data = []
if sourcetype == "WinEventLog:Security" or sourcetype == "WinEventLog":
data_manipulation = DataManipulation()
modified_file = data_manipulation.manipulate_timestamp(file_path, 'xmlwineventlog', 'WinEventLog:Security')
date_rex = r'\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2} [AP]M'
count = len(open(modified_file).readlines())
i = 0
file = fileinput.input(files=modified_file)
event = file[0]
start_position = 0
for i in range(1, count):
line = file[i]
i = i + 1
if re.match(date_rex, line):
data.append(event)
start_position = i
event = line
else:
event = event + line
data.append(event)
fileinput.close()
elif sourcetype == "xmlwineventlog":
for line in fileinput.input(files=file_path):
data.append(line)
return data
def return_macros(env):
if env == "playground":
return PULSAR_SOURCE_CONNECTION_ID_PLAYGROUND, PULSAR_SOURCE_TOPIC_PLAYGROUND
else:
return PULSAR_SOURCE_CONNECTION_ID_STAGING, PULSAR_SOURCE_TOPIC_STAGING
@@ -1,25 +0,0 @@
attrs==20.3.0
certifi==2020.12.5
chardet==4.0.0
coloredlogs==14.0
gitdb==4.0.5
GitPython==3.1.14
idna==2.10
iniconfig==1.1.1
install==1.3.4
packaging==20.9
pluggy==0.13.1
py==1.10.0
pyparsing==2.4.7
pytest==6.2.2
pytest-check==1.0.1
pytest-expect==1.1.0
pytz==2021.1
PyYAML==5.4.1
requests==2.25.1
six==1.15.0
sly==0.4
smmap==3.0.5
toml==0.10.2
u-msgpack-python==2.7.1
urllib3==1.26.5
@@ -1,100 +0,0 @@
import os
import sys
import argparse
import logging
from modules.github_service import GithubService
from modules.test_ssa_detections import SSADetectionTesting
from modules.security_content_handler import prepare_test, remove_attack_data, remove_security_content
# Logger
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
LOGGER = logging.getLogger(__name__)
def main(args):
parser = argparse.ArgumentParser(description="SSA detection smoke test")
parser.add_argument("-t", "--token", required=True,
help="specify the scloud token")
parser.add_argument("-e", "--env", required=True,
help="specify the environment")
parser.add_argument("-s", "--tenant", required=True,
help="specify the tenant in the environment")
parser.add_argument("-b", "--branch", required=True,
help="specify the security content branch")
parser.add_argument("-tf", "--test_file", required=False,
help="specify the path to the ssa test file")
parser.add_argument("-f", "--fast", required=False, default=False, action='store_true',
help="skips testing of SSA and DSP availability")
args = parser.parse_args()
token = args.token
env = args.env
tenant = args.tenant
branch = args.branch
test_file = args.test_file
fast = args.fast
# Retrieve Security Content
github_service = GithubService(branch)
ci_changes = False
if test_file:
if not os.path.isfile('security_content/tests/' + test_file):
LOGGER.error('Can not find specified test file')
sys.exit(1)
test_files_ssa = [str("tests/" + test_file)]
else:
test_files_ssa, ci_changes = github_service.get_changed_test_files_ssa()
LOGGER.info('changed/added GitHub files:')
for test_file in test_files_ssa:
LOGGER.info(test_file)
if ci_changes:
LOGGER.info('CI code has changed')
if len(test_files_ssa) == 0 and not ci_changes:
LOGGER.info('Nothing to test for SSA smoke test.')
sys.exit(0)
if len(test_files_ssa) == 0 and ci_changes and fast:
LOGGER.error('Cannot test CI changes since no detections/tests have changed and running in fast mode.')
sys.exit(-1)
# test DSP and SSA pipeline
ssa_detection_testing = SSADetectionTesting(env, tenant, token)
if not fast:
test_result_passed = ssa_detection_testing.test_dsp_pipeline()
# disabled for debugging
# if not test_result_passed:
# sys.exit(1)
# test SSA detections
test_results = []
test_passed = True
for test_file in test_files_ssa:
test_obj, attack_data_folder = prepare_test(test_file)
test_result = ssa_detection_testing.test_ssa_detections(test_obj)
test_results.append(test_result.copy())
remove_attack_data(attack_data_folder)
LOGGER.info('-----------------------------------')
LOGGER.info('------- test SSA detections -------')
LOGGER.info('-----------------------------------')
for test_result in test_results:
test_passed = test_passed and test_result['result']
LOGGER.info(test_result['msg'])
LOGGER.info('-----------------------------------')
if github_service.created_repo:
remove_security_content()
exit_code = not test_passed
sys.exit(exit_code)
if __name__ == "__main__":
main(sys.argv[1:])
@@ -1,9 +0,0 @@
#!/bin/bash
export VAULT_TOKEN=`cat /vault/.vault-token`
SCLOUD_TOKEN=$(vault read -format json $SMOKETEST_VAULT_READ_PATH tenant=$TENANT | jq -r '.data.token')
virtualenv -p python3 smoketest && source smoketest/bin/activate && pip3 install -r requirements.txt
python run_ssa_smoketest.py -t $SCLOUD_TOKEN -e $DSP_ENV -s $TENANT -b $SRCBRANCH

Some files were not shown because too many files have changed in this diff Show More