Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from chalice import Chalice
import yaml
from io import BytesIO
import cgi
app = Chalice(app_name='dvfaas-insecure-deserialization')
def _get_parts():
rfile = BytesIO(app.current_request.raw_body)
content_type = app.current_request.headers['content-type']
_, parameters = cgi.parse_header(content_type)
parameters['boundary'] = parameters['boundary'].encode('utf-8')
parsed = cgi.parse_multipart(rfile, parameters)
return parsed
@app.route('/test')
def test_route():
print('Test is working')
return {'success': 'test'}
from chalice import Chalice, Response
import time
import boto3
app = Chalice(app_name='timercheck')
home_url = 'https://alestic.com/2015/07/timercheck-scheduled-events-monitoring/'
table_name = 'timer' #TBD: Convert to chalice environment variable
def error(status, message):
return Response(
status_code=status,
body={'errorMessage': message},
)
def redirect(url):
return Response(
status_code=301,
headers={'Location': url},
body='',
)
import os
import boto3
from chalice import Chalice
from chalice import NotFoundError
from chalicelib import db
from chalicelib import rekognition
app = Chalice(app_name='media-query')
_MEDIA_DB = None
_REKOGNITION_CLIENT = None
_SUPPORTED_IMAGE_EXTENSIONS = (
'.jpg',
'.png',
)
def get_media_db():
global _MEDIA_DB
if _MEDIA_DB is None:
_MEDIA_DB = db.DynamoMediaDB(
boto3.resource('dynamodb').Table(
os.environ['MEDIA_TABLE_NAME']))
return _MEDIA_DB
from chalice import Chalice
app = Chalice(app_name='media-query')
@app.route('/')
def index():
return {'hello': 'world'}
import logging
import os
from chalice import Chalice, BadRequestError, NotFoundError
from gremlin_python import statics
from gremlin_python.structure.graph import Graph
from gremlin_python.process.graph_traversal import __
from gremlin_python.process.strategies import *
from gremlin_python.process.traversal import T, P, Operator
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
app = Chalice(app_name='neptunedemochalice')
app.debug = True
logging.basicConfig()
logging.getLogger().setLevel(logging.INFO)
def setup_graph():
try:
graph = Graph()
connstring = os.environ.get('GRAPH_DB')
logging.info('Trying To Login')
g = graph.traversal().withRemote(DriverRemoteConnection(connstring, 'g'))
logging.info('Successfully Logged In')
except Exception as e: # Shouldn't really be so broad
logging.error(e, exc_info=True)
raise BadRequestError('Could not connect to Neptune')
# mysql --ssl-mode=REQUIRED --tls-version=TLSv1.2 --ssl-ca rds-ca-2015-root.pem --host fogdog-cluster.cluster-cepyx09zrohr.us-west-2.rds.amazonaws.com --user fogdog --password
import os, ssl, subprocess
cafile = os.path.join(os.path.dirname(__file__), "chalicelib", "rds-ca-2015-root.pem")
ssl_ctx = ssl.create_default_context(cafile=cafile)
from chalice import Chalice
from sqlalchemy import create_engine
connect_args = dict(host="fogdog-cluster.cluster-cepyx09zrohr.us-west-2.rds.amazonaws.com",
db="fogdog", user="fogdog", password="fogdogfogdog", ssl=ssl_ctx)
engine = create_engine("mysql+pymysql://", connect_args=connect_args, echo=True)
app = Chalice(app_name='fogdog-api')
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(64))
fullname = Column(String(256))
password = Column(String(256))
def __repr__(self):
return "" % (
self.name, self.fullname, self.password)
import json
import os
import boto3
from chalice import Chalice
from chalice import NotFoundError
from chalicelib import db
from chalicelib import rekognition
app = Chalice(app_name='media-query')
_MEDIA_DB = None
_REKOGNITION_CLIENT = None
_SUPPORTED_IMAGE_EXTENSIONS = (
'.jpg',
'.png',
)
_SUPPORTED_VIDEO_EXTENSIONS = (
'.mp4',
'.flv',
'.mov',
)
def get_media_db():
global _MEDIA_DB
This file contains the REST API and CloudWatch entry-points for the MSAM backend.
"""
import os
import boto3
from chalice import Chalice, Rate
from chalicelib import cache
import chalicelib.channels as channel_tiles
import chalicelib.cloudwatch as cloudwatch_data
import chalicelib.layout as node_layout
import chalicelib.periodic as periodic_handlers
import chalicelib.settings as msam_settings
app = Chalice(app_name='msam')
# update one region at this interval
NODE_UPDATE_RATE_MINUTES = 5
# update connections at this interval
CONNECTION_UPDATE_RATE_MINUTES = 5
# update subscribed alarm states at this interval
ALARM_UPDATE_RATE_MINUTES = 1
# update MSAM visuals from tags at this interval
TAG_UPDATE_RATE_MINUTES = 5
# table names generated by CloudFormation
ALARMS_TABLE_NAME = os.environ["ALARMS_TABLE_NAME"]
CHANNELS_TABLE_NAME = os.environ["CHANNELS_TABLE_NAME"]
"""We will terminate execution if we have this many seconds left to process the request."""
API_GATEWAY_TIMEOUT_SECONDS = 30.0
"""
This is how quickly API Gateway gives up on Lambda. This allows us to terminate the request if the lambda has a longer
timeout than API Gateway.
"""
DSS_VERSION = os.getenv('DSS_VERSION')
"""
Tag describing the version of the currently deployed DSS codebase. Generated during deployment in the form:
[--]
"""
class DSSChaliceApp(chalice.Chalice):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._override_exptime_seconds = None
def timeout_response(method: str, path: str) -> chalice.Response:
"""
Produce a chalice Response object that indicates a timeout. Stacktraces for all running threads, other than the
current thread, are provided in the response object.
"""
frames = sys._current_frames()
current_threadid = threading.get_ident()
trace_dump = {
thread_id: traceback.format_stack(frame)
for thread_id, frame in frames.items()
if thread_id != current_threadid}
# from functools import singledispatch
from chalice import Chalice
from chalicelib.contrib.db.models import Project, Run, LoadTestResponse
app = Chalice(app_name='loadlambApi')
# get_projects, accessible via API /projects GET request, will return
# a list of projects.
@app.route('/projects')
def get_projects():
project_list = list(Project.objects().all())
return project_list
# get_project_runs, accessible via API /projects/{project_slug} or
# /runs/{project_slug} GET request, will return a list of unique run slugs
# that are linked to the specified project.
@app.route('/projects/{project_slug}')
@app.route('runs/{project_slug}')
def get_project_runs(project_slug):