sqlalchemy injection test

sqla_injection_test.py

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
#
# Use this file to build your own SSCCE
# SSCCE = Short, Self Contained, Correct (Compatible) Example
# see http://sscce.org/
#
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Standard imports

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import deferred, class_mapper
from sqlalchemy import Integer, String, Text, Binary, Column, ForeignKey, DateTime
from sqlalchemy import inspect
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# You probably don't need to overwrite this
Base = declarative_base()


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Define some models that inherit from Base

class Foo(Base):
    __tablename__ = 'foo'
    id = Column(Integer, primary_key=True)

class Bar(Base):
    __tablename__ = 'bar'
    id = Column(Integer, primary_key=True)

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# we'll use sqlite+memory to handle this
# if your issue is backend specific, then the engine will need to be different

if False:
    engine = create_engine('sqlite:///:memory:', echo=True)
else:
    engine = create_engine('postgresql://sqlalchemy_test:sqla@localhost/sqlalchemy_test')
Base.metadata.create_all(engine)


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# you shouldn't change these 2 line

sessionFactory = sessionmaker(bind=engine)
s = sessionFactory()

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Start your example here:

for i in range(0, 100):
    a = Foo(id=i)
    s.add(a)
    s.commit()

for i in range(0, 100):
    a = Bar(id=i)
    s.add(a)
    s.commit()

print "There are %s Foo" % s.query(Foo).count()
print "There are %s Bar" % s.query(Bar).count()

try:
    r = s.query(Foo).filter(sqlalchemy.text('id=1')).order_by(sqlalchemy.text('id desc; delete * from foo;')).all()
except:
    pass
s.commit()

try:
    r = s.query(Foo).filter(sqlalchemy.text('id=1')).order_by(sqlalchemy.text('id desc; delete * from bar;')).all()
except:
    pass
s.commit()

print "There are %s Foo" % s.query(Foo).count()
print "There are %s Bar" % s.query(Bar).count()



# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -