diff --git a/src/ocspdash/__init__.py b/src/ocspdash/__init__.py index 2c3b0f6..b30bd9f 100644 --- a/src/ocspdash/__init__.py +++ b/src/ocspdash/__init__.py @@ -8,6 +8,7 @@ __title__ = 'OCSPdash' # keep the __description__ synchronized with the package docstring __description__ = "A dashboard for the status of the top certificate authorities' OCSP responders." + __url__ = 'https://github.com/scolby33/OCSPdash' __author__ = 'Scott Colby' diff --git a/src/ocspdash/cli.py b/src/ocspdash/cli.py index 6929ec9..25c4f74 100644 --- a/src/ocspdash/cli.py +++ b/src/ocspdash/cli.py @@ -26,13 +26,17 @@ def web(host, port, flask_debug, verbose): logging.basicConfig(level=(logging.DEBUG if verbose else logging.INFO)) from ocspdash.web import create_application + app = create_application() app.run(host=host, port=port, debug=flask_debug) @main.command() @click.option('-n', '--buckets', default=2, type=int, help='Number of top authorities') -@click.option('--connection', help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}') +@click.option( + '--connection', + help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}', +) @click.option('-v', '--verbose', is_flag=True, help='Verbose output') def update(buckets, connection, verbose): """Update the local database.""" @@ -43,7 +47,10 @@ def update(buckets, connection, verbose): @main.command() -@click.option('--connection', help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}') +@click.option( + '--connection', + help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}', +) @click.option('-y', '--yes', is_flag=True) def nuke(connection, yes): """Nuke the database.""" @@ -53,7 +60,10 @@ def nuke(connection, yes): @main.command() -@click.option('--connection', help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}') +@click.option( + '--connection', + help=f'SQLAlchemy connection. Defaults to {OCSPDASH_DEFAULT_CONNECTION}', +) @click.argument('location_name') def new_location(connection, location_name): """Register a new location.""" @@ -61,7 +71,11 @@ def new_location(connection, location_name): invite_id, invite_validator = m.create_location(location_name) - click.echo(base64.urlsafe_b64encode(b''.join((invite_id, invite_validator))).decode("utf-8")) + click.echo( + base64.urlsafe_b64encode(b''.join((invite_id, invite_validator))).decode( + 'utf-8' + ) + ) if __name__ == '__main__': diff --git a/src/ocspdash/constants.py b/src/ocspdash/constants.py index b7a62da..89f31e5 100644 --- a/src/ocspdash/constants.py +++ b/src/ocspdash/constants.py @@ -26,19 +26,27 @@ #: The directory in which data for OCSP Dashboard is stored. Can be set from the environment variable #: ``OCSPDASH_DIRECTORY`` or defaults to ``~/.ocspdash`` -OCSPDASH_DIRECTORY = os.environ.get('OCSPDASH_DIRECTORY', os.path.join(os.path.expanduser('~'), '.ocspdash')) +OCSPDASH_DIRECTORY = os.environ.get( + 'OCSPDASH_DIRECTORY', os.path.join(os.path.expanduser('~'), '.ocspdash') +) if not os.path.exists(OCSPDASH_DIRECTORY): os.makedirs(OCSPDASH_DIRECTORY) -OCSPDASH_DEFAULT_CONNECTION = 'sqlite:///' + os.path.join(OCSPDASH_DIRECTORY, 'ocspdash.db') +OCSPDASH_DEFAULT_CONNECTION = 'sqlite:///' + os.path.join( + OCSPDASH_DIRECTORY, 'ocspdash.db' +) OCSPDASH_CONNECTION = os.environ.get('OCSPDASH_CONNECTION', OCSPDASH_DEFAULT_CONNECTION) #: The rate limit for connecting to Censys. Can be set from the environmental variable ``OCSPDASH_RATE`` or defaults # to ``0.2``. -CENSYS_RATE_LIMIT = float(os.environ.get('OCSPDASH_RATE', 0.2)) # max requests per second +CENSYS_RATE_LIMIT = float( + os.environ.get('OCSPDASH_RATE', 0.2) +) # max requests per second OCSPDASH_USER_AGENT_IDENTIFIER = f'OCSPdash/{VERSION}' -OCSPDASH_USER_AGENT = ' '.join([requests.utils.default_user_agent(), OCSPDASH_USER_AGENT_IDENTIFIER]) +OCSPDASH_USER_AGENT = ' '.join( + [requests.utils.default_user_agent(), OCSPDASH_USER_AGENT_IDENTIFIER] +) OCSP_JWT_ALGORITHM = os.environ.get('OCSP_JWT_ALGORITHM', 'ES512') diff --git a/src/ocspdash/custom_columns.py b/src/ocspdash/custom_columns.py index 242b908..bd1d5e2 100644 --- a/src/ocspdash/custom_columns.py +++ b/src/ocspdash/custom_columns.py @@ -7,9 +7,7 @@ import sqlalchemy.dialects.postgresql from sqlalchemy.types import BINARY, TypeDecorator -__all__ = [ - 'UUID', -] +__all__ = ['UUID'] class UUID(TypeDecorator): diff --git a/src/ocspdash/manager.py b/src/ocspdash/manager.py index 23d1592..db534b2 100644 --- a/src/ocspdash/manager.py +++ b/src/ocspdash/manager.py @@ -16,18 +16,22 @@ from sqlalchemy.engine import Engine from sqlalchemy.orm import scoped_session, sessionmaker -from ocspdash.constants import OCSPDASH_DEFAULT_CONNECTION, OCSPDASH_USER_AGENT_IDENTIFIER +from ocspdash.constants import ( + OCSPDASH_DEFAULT_CONNECTION, + OCSPDASH_USER_AGENT_IDENTIFIER, +) from ocspdash.models import Authority, Base, Chain, Location, Responder, Result from ocspdash.security import pwd_context from ocspdash.server_query import ServerQuery -__all__ = [ - 'Manager', -] +__all__ = ['Manager'] logger = logging.getLogger(__name__) -ManifestEntry = namedtuple('ManifestEntry', 'responder_url subject_certificate issuer_certificate chain_certificate_hash') +ManifestEntry = namedtuple( + 'ManifestEntry', + 'responder_url subject_certificate issuer_certificate chain_certificate_hash', +) def _workaround_pysqlite_transaction_bug(): @@ -48,12 +52,16 @@ def do_connect(dbapi_connection, connection_record): if isinstance(dbapi_connection, _sqlite3_Connection): # disable pysqlite's emitting of the BEGIN statement entirely. # also stops it from emitting COMMIT before any DDL. - logger.debug('setting connection isolation level to `None` to work around pysqlite bug') + logger.debug( + 'setting connection isolation level to `None` to work around pysqlite bug' + ) dbapi_connection.isolation_level = None @_event.listens_for(_Engine, 'begin') def do_begin(connection): - if isinstance(connection._Connection__connection.connection, _sqlite3_Connection): + if isinstance( + connection._Connection__connection.connection, _sqlite3_Connection + ): # emit our own BEGIN logger.debug('emitting our own BEGIN to work around pysqlite bug') connection.execute('BEGIN') @@ -65,7 +73,12 @@ def do_begin(connection): class Manager(object): """Manager for interacting with the database.""" - def __init__(self, engine: Engine, session: scoped_session, server_query: Optional[ServerQuery] = None) -> None: + def __init__( + self, + engine: Engine, + session: scoped_session, + server_query: Optional[ServerQuery] = None, + ) -> None: """Instantiate a Manager with instances of the objects it needs. :param engine: The database engine. @@ -79,7 +92,13 @@ def __init__(self, engine: Engine, session: scoped_session, server_query: Option self.create_all() @classmethod - def from_args(cls, connection: Optional[str] = None, echo: bool = False, api_id: Optional[str] = None, api_secret: Optional[str] = None) -> 'Manager': + def from_args( + cls, + connection: Optional[str] = None, + echo: bool = False, + api_id: Optional[str] = None, + api_secret: Optional[str] = None, + ) -> 'Manager': """Instantiate a Manager along with the objects it needs. :param connection: An SQLAlchemy-compatible connection string. @@ -89,7 +108,9 @@ def from_args(cls, connection: Optional[str] = None, echo: bool = False, api_id: :returns: An instance of Manager configured according to the arguments provided. """ - engine, session = cls._get_engine_from_connection(connection=connection, echo=echo) + engine, session = cls._get_engine_from_connection( + connection=connection, echo=echo + ) server_query = cls._get_server_query(api_id=api_id, api_secret=api_secret) @@ -115,7 +136,9 @@ def _get_connection(connection: Optional[str] = None): return OCSPDASH_DEFAULT_CONNECTION @staticmethod - def _get_credentials(user: Optional[str] = None, password: Optional[str] = None) -> Tuple[Optional[str], Optional[str]]: + def _get_credentials( + user: Optional[str] = None, password: Optional[str] = None + ) -> Tuple[Optional[str], Optional[str]]: if user is None: user = os.environ.get('CENSYS_API_ID') @@ -125,24 +148,30 @@ def _get_credentials(user: Optional[str] = None, password: Optional[str] = None) return user, password @classmethod - def _get_engine_from_connection(cls, connection: Optional[str] = None, echo: bool = False) -> Tuple[Engine, scoped_session]: + def _get_engine_from_connection( + cls, connection: Optional[str] = None, echo: bool = False + ) -> Tuple[Engine, scoped_session]: connection = cls._get_connection(connection) engine = create_engine(connection, echo=echo) - session_maker = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False) + session_maker = sessionmaker( + bind=engine, autoflush=False, expire_on_commit=False + ) session = scoped_session(session_maker) return engine, session @classmethod - def _get_server_query(cls, api_id: Optional[str] = None, api_secret: Optional[str] = None) -> Optional[ServerQuery]: + def _get_server_query( + cls, api_id: Optional[str] = None, api_secret: Optional[str] = None + ) -> Optional[ServerQuery]: api_id, api_secret = cls._get_credentials(user=api_id, password=api_secret) if api_id is not None and api_secret is not None: return ServerQuery( api_id=api_id, api_secret=api_secret, - user_agent_identifier=OCSPDASH_USER_AGENT_IDENTIFIER + user_agent_identifier=OCSPDASH_USER_AGENT_IDENTIFIER, ) def create_all(self, checkfirst=True): @@ -175,7 +204,9 @@ def get_authority_by_name(self, name: str) -> Optional[Authority]: :returns: The Authority or None """ - return self.session.query(Authority).filter(Authority.name == name).one_or_none() + return ( + self.session.query(Authority).filter(Authority.name == name).one_or_none() + ) def ensure_authority(self, name: str, cardinality: int) -> Authority: """Create or update an Authority in the DB. @@ -191,10 +222,7 @@ def ensure_authority(self, name: str, cardinality: int) -> Authority: authority = self.get_authority_by_name(name) if authority is None: - authority = Authority( - name=name, - cardinality=cardinality, - ) + authority = Authority(name=name, cardinality=cardinality) self.session.add(authority) else: @@ -215,7 +243,9 @@ def get_responder(self, authority: Authority, url: str) -> Optional[Responder]: f = and_(Responder.authority_id == authority.id, Responder.url == url) return self.session.query(Responder).filter(f).one_or_none() - def ensure_responder(self, authority: Authority, url: str, cardinality: int) -> Responder: + def ensure_responder( + self, authority: Authority, url: str, cardinality: int + ) -> Responder: """Create or update a responder in the DB. If a responder with the given Authority and URL exists, the cardinality is updated. @@ -230,11 +260,7 @@ def ensure_responder(self, authority: Authority, url: str, cardinality: int) -> responder = self.get_responder(authority=authority, url=url) if responder is None: - responder = Responder( - authority=authority, - url=url, - cardinality=cardinality - ) + responder = Responder(authority=authority, url=url, cardinality=cardinality) self.session.add(responder) else: @@ -251,17 +277,27 @@ def get_chain_by_certificate_hash(self, certificate_hash: bytes) -> Optional[Cha :returns: the Chain or None """ - return self.session.query(Chain).filter(Chain.certificate_hash == certificate_hash).one_or_none() + return ( + self.session.query(Chain) + .filter(Chain.certificate_hash == certificate_hash) + .one_or_none() + ) - def get_most_recent_chain_by_responder(self, responder: Responder) -> Optional[Chain]: + def get_most_recent_chain_by_responder( + self, responder: Responder + ) -> Optional[Chain]: """Get the newest chain for a Responder. :param responder: the Responder whose chain we're seeking :returns: the Chain or None """ - return self.session.query(Chain).filter(Chain.responder_id == responder.id).order_by( - Chain.retrieved.desc()).first() + return ( + self.session.query(Chain) + .filter(Chain.responder_id == responder.id) + .order_by(Chain.retrieved.desc()) + .first() + ) def ensure_chain(self, responder: Responder) -> Optional[Chain]: """Get or create a chain for a Responder. @@ -289,16 +325,14 @@ def ensure_chain(self, responder: Responder) -> Optional[Chain]: if not responder.current: return most_recent_chain - subject, issuer = self.server_query.get_certs_for_issuer_and_url(responder.authority.name, responder.url) + subject, issuer = self.server_query.get_certs_for_issuer_and_url( + responder.authority.name, responder.url + ) if subject is None or issuer is None: return None - chain = Chain( - responder=responder, - subject=subject, - issuer=issuer, - ) + chain = Chain(responder=responder, subject=subject, issuer=issuer) self.session.add(chain) self.session.commit() @@ -332,8 +366,9 @@ def update(self, n: int = 10): raise RuntimeError('No username and password for Censys supplied') authorities = self.get_top_authorities(n) - if (not authorities or # probably a first run with a clean DB - any(authority.old for authority in authorities)): + if not authorities or any( # probably a first run with a clean DB + authority.old for authority in authorities + ): issuers = self.server_query.get_top_authorities(buckets=n) for issuer_name, issuer_cardinality in issuers.items(): authority = self.ensure_authority(issuer_name, issuer_cardinality) @@ -341,7 +376,9 @@ def update(self, n: int = 10): ocsp_urls = self.server_query.get_ocsp_urls_for_issuer(authority.name) for url, responder_cardinality in ocsp_urls.items(): - responder = self.ensure_responder(authority, url, responder_cardinality) + responder = self.ensure_responder( + authority, url, responder_cardinality + ) self.ensure_chain(responder) authorities = self.get_top_authorities(n) @@ -362,23 +399,32 @@ def get_top_authorities(self, n: int = 10) -> List[Authority]: :returns: a list of up to n Authorities """ - return self.session.query(Authority).order_by(Authority.cardinality.desc()).limit(n).all() + return ( + self.session.query(Authority) + .order_by(Authority.cardinality.desc()) + .limit(n) + .all() + ) - def get_most_recent_result_for_each_location(self) -> List[Tuple[Authority, Responder, Result, Location]]: + def get_most_recent_result_for_each_location( + self + ) -> List[Tuple[Authority, Responder, Result, Location]]: """Get the most recent results for each location.""" - return self.session.query(Authority, Responder, Result, Location) \ - .join(Responder) \ - .join(Chain) \ - .join(Result) \ - .join(Location) \ - .group_by(Responder, Location) \ - .having(func.max(Result.retrieved)) \ - .order_by(Authority.cardinality.desc()) \ - .order_by(Authority.name) \ - .order_by(Responder.cardinality.desc()) \ - .order_by(Responder.url) \ - .order_by(Location.name) \ + return ( + self.session.query(Authority, Responder, Result, Location) + .join(Responder) + .join(Chain) + .join(Result) + .join(Location) + .group_by(Responder, Location) + .having(func.max(Result.retrieved)) + .order_by(Authority.cardinality.desc()) + .order_by(Authority.name) + .order_by(Responder.cardinality.desc()) + .order_by(Responder.url) + .order_by(Location.name) .all() + ) def get_all_locations_with_test_results(self) -> List[Location]: """Return all the Location objects that have at least one associated Result.""" @@ -393,24 +439,22 @@ def get_payload(self): locations = self.get_all_locations_with_test_results() sections = OrderedDict() - for authority, group in groupby(self.get_most_recent_result_for_each_location(), itemgetter(0)): + for authority, group in groupby( + self.get_most_recent_result_for_each_location(), itemgetter(0) + ): sections[authority.name] = [] for responder, group2 in groupby(group, itemgetter(1)): - results = tuple( - result - for _, _, result, _ in group2 - ) + results = tuple(result for _, _, result, _ in group2) row = (responder.url, responder.current) + results sections[authority.name].append(row) - return { - 'locations': locations, - 'sections': sections - } + return {'locations': locations, 'sections': sections} def get_location_by_key_id(self, key_id: uuid.UUID) -> Optional[Location]: """Get a location by its key id.""" - return self.session.query(Location).filter(Location.key_id == key_id).one_or_none() + return ( + self.session.query(Location).filter(Location.key_id == key_id).one_or_none() + ) def create_location(self, location_name: str) -> Tuple[bytes, bytes]: """Create a new Location with an invite. @@ -424,9 +468,7 @@ def create_location(self, location_name: str) -> Tuple[bytes, bytes]: invite_validator_hash = pwd_context.hash(validator) new_location = Location( - name=location_name, - selector=selector, - validator_hash=invite_validator_hash + name=location_name, selector=selector, validator_hash=invite_validator_hash ) self.session.add(new_location) @@ -435,9 +477,15 @@ def create_location(self, location_name: str) -> Tuple[bytes, bytes]: def get_location_by_selector(self, selector: bytes) -> Optional[Location]: """Get an invite by its binary selector.""" - return self.session.query(Location).filter(Location.selector == selector).one_or_none() + return ( + self.session.query(Location) + .filter(Location.selector == selector) + .one_or_none() + ) - def process_location(self, invite_token: bytes, public_key: str) -> Optional[Location]: + def process_location( + self, invite_token: bytes, public_key: str + ) -> Optional[Location]: """Given an invite token and public key, check for a valid invite and associate the public key with the corresponding location. :parameter invite_token: a 32-byte string corresponding to an invited Location. @@ -464,10 +512,11 @@ def process_location(self, invite_token: bytes, public_key: str) -> Optional[Loc return location def _get_top_authorities_responders(self) -> List[Responder]: - return list(itt.chain.from_iterable(( - authority.responders - for authority in self.get_top_authorities() - ))) + return list( + itt.chain.from_iterable( + (authority.responders for authority in self.get_top_authorities()) + ) + ) def _get_manifest_chains(self) -> List[Chain]: responders = self._get_top_authorities_responders() @@ -476,11 +525,16 @@ def _get_manifest_chains(self) -> List[Chain]: chains = [ responder.most_recent_chain for responder in responders - if responder.most_recent_chain is not None] + if responder.most_recent_chain is not None + ] if len(responders) != len(chains): # TODO why is this needed? Originally it was an assertion... - logger.warning('Number of responders and number of chains mismatch: %d responders and %d chains', len(responders), len(chains)) + logger.warning( + 'Number of responders and number of chains mismatch: %d responders and %d chains', + len(responders), + len(chains), + ) return chains diff --git a/src/ocspdash/models.py b/src/ocspdash/models.py index 0ceeb2d..f7af5ca 100644 --- a/src/ocspdash/models.py +++ b/src/ocspdash/models.py @@ -11,7 +11,17 @@ from typing import Optional # noqa: F401 imported for PyCharm type checking from oscrypto import asymmetric -from sqlalchemy import Binary, Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint +from sqlalchemy import ( + Binary, + Boolean, + Column, + DateTime, + ForeignKey, + Integer, + String, + Text, + UniqueConstraint, +) from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base from sqlalchemy.orm import backref, relationship from sqlalchemy.sql import functions as func @@ -39,10 +49,15 @@ class Authority(Base): id = Column(Integer, primary_key=True) - name = Column(String(255), nullable=False, index=True, doc='the name of the authority') + name = Column( + String(255), nullable=False, index=True, doc='the name of the authority' + ) - cardinality = Column(Integer, doc="The number of certs observed from this authority in the wild. Update this " - "when rankings change. From the Censys crawler.") + cardinality = Column( + Integer, + doc='The number of certs observed from this authority in the wild. Update this ' + 'when rankings change. From the Censys crawler.', + ) last_updated = Column(DateTime, server_default=func.now(), onupdate=func.now()) @@ -68,7 +83,7 @@ def to_json(self): 'current': responder.current, } for responder in self.responders - ] + ], } @@ -79,19 +94,22 @@ class Responder(Base): id = Column(Integer, primary_key=True) - authority_id = Column(Integer, ForeignKey('authority.id'), nullable=False, doc='the authority') + authority_id = Column( + Integer, ForeignKey('authority.id'), nullable=False, doc='the authority' + ) authority = relationship('Authority', backref=backref('responders')) url = Column(Text, nullable=False, doc='the URL of the OCSP endpoint') - cardinality = Column(Integer, doc="The number of certs observed using this authority/endpoint pair in the " - "wild. Update this when rankings are updated.") + cardinality = Column( + Integer, + doc='The number of certs observed using this authority/endpoint pair in the ' + 'wild. Update this when rankings are updated.', + ) last_updated = Column(DateTime, server_default=func.now(), onupdate=func.now()) - __table_args__ = ( - UniqueConstraint(authority_id, url), - ) + __table_args__ = (UniqueConstraint(authority_id, url),) def __repr__(self): return f'{self.authority} at {self.url}' @@ -99,10 +117,7 @@ def __repr__(self): @property def current(self) -> bool: """Calculate if this responder is current by the status of its most recent result over all chains.""" - return not all( - chain.expired - for chain in self.chains - ) + return not all(chain.expired for chain in self.chains) @property def most_recent_chain(self) -> 'Optional[Chain]': @@ -151,17 +166,28 @@ class Chain(Base): responder = relationship('Responder', backref=backref('chains')) subject = Column(Binary, nullable=False, doc='raw bytes of the subject certificate') - issuer = Column(Binary, nullable=False, doc="raw bytes of the subject's issuer certificate") - retrieved = Column(DateTime, default=datetime.utcnow, nullable=False, - doc='expire the cached chain when this date is more than 7 days ago') - - certificate_hash = Column(Binary(64), nullable=False, unique=True, default=_certificate_hash_default, onupdate=_certificate_hash_default, index=True, - doc='') + issuer = Column( + Binary, nullable=False, doc="raw bytes of the subject's issuer certificate" + ) + retrieved = Column( + DateTime, + default=datetime.utcnow, + nullable=False, + doc='expire the cached chain when this date is more than 7 days ago', + ) - __table_args__ = ( - UniqueConstraint(subject, issuer), + certificate_hash = Column( + Binary(64), + nullable=False, + unique=True, + default=_certificate_hash_default, + onupdate=_certificate_hash_default, + index=True, + doc='', ) + __table_args__ = (UniqueConstraint(subject, issuer),) + @property def expired(self) -> bool: """Return True if the subject certificate has expired, False otherwise.""" @@ -238,7 +264,8 @@ def __repr__(self): return f'Invite for {self.name}' def to_json( - self): # TODO: this gets returned by the /register endpoint and some of this info could be considered private. Should it be removed from the to_json? + self + ): # TODO: this gets returned by the /register endpoint and some of this info could be considered private. Should it be removed from the to_json? """Return a representation of the instance suitable for passing in to JSON conversion.""" return { 'id': self.id, @@ -247,9 +274,7 @@ def to_json( 'validator_hash': self.validator_hash, 'pubkey': str(self.pubkey), 'key_id': str(self.key_id), - 'results': [ - result.id for result in self.results - ] + 'results': [result.id for result in self.results], } @@ -260,16 +285,27 @@ class Result(Base): id = Column(Integer, primary_key=True) - chain_id = Column(Integer, ForeignKey('chain.id'), doc='the certificate chain that was used for the OCSP test') + chain_id = Column( + Integer, + ForeignKey('chain.id'), + doc='the certificate chain that was used for the OCSP test', + ) chain = relationship('Chain', backref=backref('results')) - location_id = Column(Integer, ForeignKey('location.id'), nullable=False, doc='the location that ran the test') + location_id = Column( + Integer, + ForeignKey('location.id'), + nullable=False, + doc='the location that ran the test', + ) location = relationship('Location', backref=backref('results', lazy='dynamic')) retrieved = Column(DateTime, default=datetime.utcnow, doc='when the test was run') ping = Column(Boolean, nullable=False, doc='did the server respond to a ping?') - ocsp = Column(Boolean, nullable=False, doc='did a valid OCSP request get a good response?') + ocsp = Column( + Boolean, nullable=False, doc='did a valid OCSP request get a good response?' + ) @property def status(self) -> OCSPResponderStatus: # relates to the glyphicon displayed @@ -292,10 +328,7 @@ def to_json(self): """Return a representation of the instance suitable for passing in to JSON conversion.""" return { 'id': self.id, - 'location': { - 'id': self.location.id, - 'location': self.location.name - }, + 'location': {'id': self.location.id, 'location': self.location.name}, 'chain': { 'id': self.chain.id, 'retrieved': str(self.chain.retrieved), diff --git a/src/ocspdash/ocspscrape.py b/src/ocspdash/ocspscrape.py index b134cf4..82964ab 100644 --- a/src/ocspdash/ocspscrape.py +++ b/src/ocspdash/ocspscrape.py @@ -90,10 +90,7 @@ def main(): print(f'public key:\t{claims["pk"]}'.expandtabs(7)[:78] + '..') print(f'invite token:\t{claims["token"]}'.expandtabs(7)) else: - token = scrape( - json.loads(line) - for line in tqdm(sys.stdin) - ) + token = scrape(json.loads(line) for line in tqdm(sys.stdin)) print(token) @@ -103,17 +100,16 @@ def genkey(invite_token: str): serialized_private_key = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() + encryption_algorithm=serialization.NoEncryption(), + ).decode('utf-8') + public_key = b64encode( + private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) ).decode('utf-8') - public_key = b64encode(private_key.public_key().public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - )).decode('utf-8') - payload = { - 'pk': public_key, - 'token': invite_token - } + payload = {'pk': public_key, 'token': invite_token} token = jwt.encode(payload, private_key, algorithm=JWT_ALGORITHM) return token, serialized_private_key @@ -131,7 +127,13 @@ def scrape(queries): """Scrape the OCSP responders provided.""" # TODO needs type hint for return requests_session = requests.Session() - requests_session.headers.update({'User-Agent': ' '.join([requests.utils.default_user_agent(), 'OCSPscrape 0.1.0'])}) + requests_session.headers.update( + { + 'User-Agent': ' '.join( + [requests.utils.default_user_agent(), 'OCSPscrape 0.1.0'] + ) + } + ) payload = {RESULTS_JWT_CLAIM: []} @@ -146,16 +148,20 @@ def scrape(queries): time = datetime.utcnow().strftime('%FT%TZ') ping_result = ping(netloc) - ocsp_result = check_ocsp_response(subject_bytes, issuer_bytes, responder_url, requests_session) + ocsp_result = check_ocsp_response( + subject_bytes, issuer_bytes, responder_url, requests_session + ) - payload[RESULTS_JWT_CLAIM].append({ - # 'authority_name': authority_name, - 'chain_certificate_hash': query['chain_certificate_hash'], - # 'responder_url': responder_url, - 'time': time, - 'ping': ping_result, - 'ocsp': ocsp_result - }) + payload[RESULTS_JWT_CLAIM].append( + { + # 'authority_name': authority_name, + 'chain_certificate_hash': query['chain_certificate_hash'], + # 'responder_url': responder_url, + 'time': time, + 'ping': ping_result, + 'ocsp': ocsp_result, + } + ) # TODO handle missing env vars more gracefully key = os.environ['OCSPSCRAPE_PRIVATE_KEY'] @@ -167,14 +173,14 @@ def scrape(queries): def _keyid_from_private_key(private_key_data: str) -> uuid.UUID: loaded_private_key = serialization.load_pem_private_key( - data=private_key_data.encode('utf-8'), - password=None, - backend=default_backend() + data=private_key_data.encode('utf-8'), password=None, backend=default_backend() ) - public_key = b64encode(loaded_private_key.public_key().public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - )).decode('utf-8') + public_key = b64encode( + loaded_private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ).decode('utf-8') key_id = uuid.uuid5(NAMESPACE_OCSPDASH_KID, public_key) return key_id @@ -191,7 +197,9 @@ def ping(host: str) -> bool: return results.returncode == 0 -def check_ocsp_response(subject_cert: bytes, issuer_cert: bytes, url: str, session: requests.Session) -> bool: +def check_ocsp_response( + subject_cert: bytes, issuer_cert: bytes, url: str, session: requests.Session +) -> bool: """Create and send an OCSP request. :param subject_cert: The certificate that information is being requested about @@ -211,7 +219,11 @@ def check_ocsp_response(subject_cert: bytes, issuer_cert: bytes, url: str, sessi ocsp_request = builder.build() try: - ocsp_resp = session.post(url, data=ocsp_request.dump(), headers={'Content-Type': 'application/ocsp-request'}) + ocsp_resp = session.post( + url, + data=ocsp_request.dump(), + headers={'Content-Type': 'application/ocsp-request'}, + ) except requests.RequestException: return False @@ -220,7 +232,10 @@ def check_ocsp_response(subject_cert: bytes, issuer_cert: bytes, url: str, sessi except ValueError: return False - return parsed_ocsp_response and parsed_ocsp_response.native['response_status'] == 'successful' + return ( + parsed_ocsp_response + and parsed_ocsp_response.native['response_status'] == 'successful' + ) if __name__ == '__main__': diff --git a/src/ocspdash/security.py b/src/ocspdash/security.py index 0cf1624..6396f14 100644 --- a/src/ocspdash/security.py +++ b/src/ocspdash/security.py @@ -4,7 +4,4 @@ from passlib.context import CryptContext -pwd_context = CryptContext( - schemes=['argon2'], - deprecated='auto', -) +pwd_context = CryptContext(schemes=['argon2'], deprecated='auto') diff --git a/src/ocspdash/server_query.py b/src/ocspdash/server_query.py index db41169..8212570 100644 --- a/src/ocspdash/server_query.py +++ b/src/ocspdash/server_query.py @@ -16,20 +16,13 @@ def _get_results(report): - return sorted( - report['results'], - key=itemgetter('doc_count'), - reverse=True - ) + return sorted(report['results'], key=itemgetter('doc_count'), reverse=True) def _get_results_as_dict(report): results = _get_results(report) - return OrderedDict([ - (result['key'], result['doc_count']) - for result in results - ]) + return OrderedDict([(result['key'], result['doc_count']) for result in results]) class ServerQuery(RateLimitedCensysCertificates): @@ -45,7 +38,7 @@ def get_top_authorities(self, buckets: int = 10) -> MutableMapping[str, int]: report = self.report( query='validation.nss.valid: true', field='parsed.issuer.organization', - buckets=buckets + buckets=buckets, ) return _get_results_as_dict(report) @@ -59,7 +52,7 @@ def get_ocsp_urls_for_issuer(self, issuer: str) -> MutableMapping[str, int]: """ report = self.report( query=f'validation.nss.valid: true AND parsed.issuer.organization: "{issuer}"', - field='parsed.extensions.authority_info_access.ocsp_urls' + field='parsed.extensions.authority_info_access.ocsp_urls', ) return _get_results_as_dict(report) @@ -80,17 +73,20 @@ def is_ocsp_url_current_for_issuer(self, issuer: str, url: str) -> bool: """ tags_report = self.report( query=f'validation.nss.valid: true AND parsed.issuer.organization: "{issuer}" AND parsed.extensions.authority_info_access.ocsp_urls.raw: "{url}" AND (tags: "unexpired" OR tags: "expired")', - field='tags' + field='tags', ) results = { - result['key']: result['doc_count'] - for result in tags_report['results'] + result['key']: result['doc_count'] for result in tags_report['results'] } - return self._url_not_expired(results) # TODO: turn this return to a function to document implicitly + return self._url_not_expired( + results + ) # TODO: turn this return to a function to document implicitly - def get_certs_for_issuer_and_url(self, issuer: str, url: str) -> Union[Tuple[bytes, bytes], Tuple[None, None]]: + def get_certs_for_issuer_and_url( + self, issuer: str, url: str + ) -> Union[Tuple[bytes, bytes], Tuple[None, None]]: """Retrieve the raw bytes for an example subject certificate and its issuing cert for a given authority and OCSP url. :param issuer: The name of the authority from which a certificate is sought @@ -104,7 +100,11 @@ def get_certs_for_issuer_and_url(self, issuer: str, url: str) -> Union[Tuple[byt base_query = f'validation.nss.valid: true AND parsed.issuer.organization: "{issuer}" AND parsed.extensions.authority_info_access.ocsp_urls.raw: "{url}" AND parsed.extensions.authority_info_access.issuer_urls: /.+/' search = self.search( query=f'{base_query} AND tags: "unexpired"', - fields=['parsed.extensions.authority_info_access.issuer_urls', 'parsed.names', 'raw'] + fields=[ + 'parsed.extensions.authority_info_access.issuer_urls', + 'parsed.names', + 'raw', + ], ) subject_cert = next(search, None) if subject_cert is None: @@ -112,14 +112,20 @@ def get_certs_for_issuer_and_url(self, issuer: str, url: str) -> Union[Tuple[byt logger.info('Searching for an expired certificate instead') search = self.search( query=base_query, - fields=['parsed.extensions.authority_info_access.issuer_urls', 'parsed.names', 'raw'] + fields=[ + 'parsed.extensions.authority_info_access.issuer_urls', + 'parsed.names', + 'raw', + ], ) subject_cert = next(search, None) if subject_cert is None: return None, None logger.debug(f'Getting issuer cert for {issuer}: {url}') - issuer_urls = subject_cert['parsed.extensions.authority_info_access.issuer_urls'] + issuer_urls = subject_cert[ + 'parsed.extensions.authority_info_access.issuer_urls' + ] for issuer_url in issuer_urls: try: resp = requests_session.get(issuer_url) diff --git a/src/ocspdash/web/__init__.py b/src/ocspdash/web/__init__.py index 755a802..1505d0d 100644 --- a/src/ocspdash/web/__init__.py +++ b/src/ocspdash/web/__init__.py @@ -4,6 +4,4 @@ from .app import create_application -__all__ = [ - 'create_application' -] +__all__ = ['create_application'] diff --git a/src/ocspdash/web/admin.py b/src/ocspdash/web/admin.py index 72711d6..7a44791 100644 --- a/src/ocspdash/web/admin.py +++ b/src/ocspdash/web/admin.py @@ -8,9 +8,7 @@ from ocspdash.models import Authority, Chain, Location, Responder, Result -__all__ = [ - 'make_admin', -] +__all__ = ['make_admin'] def make_admin(app: Flask, session) -> Admin: diff --git a/src/ocspdash/web/app.py b/src/ocspdash/web/app.py index f312da6..37c4f0f 100644 --- a/src/ocspdash/web/app.py +++ b/src/ocspdash/web/app.py @@ -15,9 +15,7 @@ from ocspdash.web.blueprints import api, ui from ocspdash.web.extension import OCSPSQLAlchemy -__all__ = [ - 'create_application', -] +__all__ = ['create_application'] logger = logging.getLogger('web') @@ -25,14 +23,18 @@ def create_application() -> Flask: """Create the OCSPdash Flask application.""" app = Flask(__name__) - app.config.update(dict( - SQLALCHEMY_DATABASE_URI=os.environ.get('OCSPDASH_CONNECTION', OCSPDASH_DEFAULT_CONNECTION), - SQLALCHEMY_TRACK_MODIFICATIONS=False, - SECRET_KEY=os.environ.get('OCSPDASH_SECRET_KEY', 'test key'), - DEBUG=os.environ.get('OCSPDASH_DEBUG', False), - CENSYS_API_ID=os.environ.get('CENSYS_API_ID'), - CENSYS_API_SECRET=os.environ.get('CENSYS_API_SECRET'), - )) + app.config.update( + dict( + SQLALCHEMY_DATABASE_URI=os.environ.get( + 'OCSPDASH_CONNECTION', OCSPDASH_DEFAULT_CONNECTION + ), + SQLALCHEMY_TRACK_MODIFICATIONS=False, + SECRET_KEY=os.environ.get('OCSPDASH_SECRET_KEY', 'test key'), + DEBUG=os.environ.get('OCSPDASH_DEBUG', False), + CENSYS_API_ID=os.environ.get('CENSYS_API_ID'), + CENSYS_API_SECRET=os.environ.get('CENSYS_API_SECRET'), + ) + ) db = OCSPSQLAlchemy(app=app) Bootstrap(app) diff --git a/src/ocspdash/web/blueprints/__init__.py b/src/ocspdash/web/blueprints/__init__.py index 67cbf7d..921eb98 100644 --- a/src/ocspdash/web/blueprints/__init__.py +++ b/src/ocspdash/web/blueprints/__init__.py @@ -5,7 +5,4 @@ from .api import api from .ui import ui -__all__ = [ - 'api', - 'ui' -] +__all__ = ['api', 'ui'] diff --git a/src/ocspdash/web/blueprints/api.py b/src/ocspdash/web/blueprints/api.py index 575aa40..8478f09 100644 --- a/src/ocspdash/web/blueprints/api.py +++ b/src/ocspdash/web/blueprints/api.py @@ -61,14 +61,20 @@ def get_manifest(): 'responder_url': responder_url, 'subject_certificate': b64encode(subject_certificate).decode('utf-8'), 'issuer_certificate': b64encode(issuer_certificate).decode('utf-8'), - 'chain_certificate_hash': b64encode(chain_certificate_hash).decode('utf-8'), + 'chain_certificate_hash': b64encode(chain_certificate_hash).decode( + 'utf-8' + ), } for responder_url, subject_certificate, issuer_certificate, chain_certificate_hash in manifest_data ) - return manifest_lines.getvalue(), { - 'Content-Type': 'application/json', 'Content-Disposition': 'inline; filename="manifest.jsonl"' - } + return ( + manifest_lines.getvalue(), + { + 'Content-Type': 'application/json', + 'Content-Disposition': 'inline; filename="manifest.jsonl"', + }, + ) def _prepare_result_dictionary(result_data): @@ -81,7 +87,7 @@ def _prepare_result_dictionary(result_data): 'chain': chain, 'retrieved': retrieved, 'ping': result_data['ping'], - 'ocsp': result_data['ocsp'] + 'ocsp': result_data['ocsp'], } @@ -98,12 +104,14 @@ def submit(): except JWTError: return abort(400) - prepared_result_dicts = (_prepare_result_dictionary(result_data) - for result_data in claims['res']) + prepared_result_dicts = ( + _prepare_result_dictionary(result_data) for result_data in claims['res'] + ) manager.insert_payload(submitting_location, prepared_result_dicts) return ('', HTTPStatus.NO_CONTENT) + # @api.route('/status') # def get_payload(): # """Spits back the current payload""" diff --git a/src/ocspdash/web/blueprints/ui.py b/src/ocspdash/web/blueprints/ui.py index de85163..7d095ce 100644 --- a/src/ocspdash/web/blueprints/ui.py +++ b/src/ocspdash/web/blueprints/ui.py @@ -6,9 +6,7 @@ from ocspdash.web.proxies import manager -__all__ = [ - 'ui', -] +__all__ = ['ui'] ui = Blueprint('ui', __name__) @@ -19,6 +17,7 @@ def home(): payload = manager.get_payload() return render_template('index.html', payload=payload) + # @ui.route('/submit', methods=['POST']) # def submit(): # """Show the submit view.""" diff --git a/src/ocspdash/web/extension.py b/src/ocspdash/web/extension.py index 8637677..9c3c7d6 100644 --- a/src/ocspdash/web/extension.py +++ b/src/ocspdash/web/extension.py @@ -7,9 +7,7 @@ from ocspdash.manager import Manager -__all__ = [ - 'OCSPSQLAlchemy', -] +__all__ = ['OCSPSQLAlchemy'] class OCSPSQLAlchemy(SQLAlchemy): diff --git a/src/ocspdash/web/proxies.py b/src/ocspdash/web/proxies.py index 2983139..43acfcc 100644 --- a/src/ocspdash/web/proxies.py +++ b/src/ocspdash/web/proxies.py @@ -8,9 +8,7 @@ from ocspdash.manager import Manager from ocspdash.web.extension import OCSPSQLAlchemy -__all__ = [ - 'manager', -] +__all__ = ['manager'] def get_manager_proxy(): diff --git a/tests/conftest.py b/tests/conftest.py index 448192d..099f99a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,11 +85,7 @@ def restart_savepoint(session, transaction): session.begin_nested() logger.debug('create Manager') - manager = Manager( - engine=engine, - session=session, - server_query=None - ) + manager = Manager(engine=engine, session=session, server_query=None) logger.debug('yielding from session') yield manager, connection diff --git a/tests/constants.py b/tests/constants.py index 566af36..f753b63 100644 --- a/tests/constants.py +++ b/tests/constants.py @@ -5,12 +5,7 @@ import os import uuid -__all__ = [ - 'TEST_LOCATION_NAME', - 'TEST_PUBLIC_KEY', - 'TEST_KEY_ID', - 'TEST_CONNECTION', -] +__all__ = ['TEST_LOCATION_NAME', 'TEST_PUBLIC_KEY', 'TEST_KEY_ID', 'TEST_CONNECTION'] TEST_LOCATION_NAME = 'YOLO' TEST_PUBLIC_KEY = 'LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlHYk1CQUdCeXFHU000OUFnRUdCU3VCQkFBakE0R0dBQVFCc0orTXJLWU1OdlVPQXZnMThwd0hRTTRnMGRqbQpvaUx5WmFxeTdnQ3ZiT0FZOFo5NmxXSVV4K2NCaVJpZkJrTzlZY2M5UHBHbzA5U2E5Rlo4Z0FZTjluZ0JHR1BTCktsWjlJZUJMZWpQVlBMRk9rMmkwekxwbnVFQ1d2aFhuUE9RazFPSlo4blFOQnN2RWFndXgyRlZIQytJaFlkVVUKbFBJMU8rRzVmTHZ5ZnVnNTBBND0KLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg==' diff --git a/tests/test_manager.py b/tests/test_manager.py index ebc1a23..6d1fba3 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -10,15 +10,13 @@ def test_ensure_authority(manager_function: Manager): """Test the creation of Authority objects.""" authority1 = manager_function.ensure_authority( - name='Test Authority', - cardinality=1234 + name='Test Authority', cardinality=1234 ) assert authority1.name == 'Test Authority' assert authority1.cardinality == 1234 authority2 = manager_function.ensure_authority( - name='Test Authority', - cardinality=2345 + name='Test Authority', cardinality=2345 ) assert authority1 is authority2 assert authority2.name == 'Test Authority' @@ -37,7 +35,9 @@ def test_location_invites(manager_function: Manager): assert location.pubkey is None assert location.key_id is None - processed_location = manager_function.process_location(b''.join((selector, validator)), TEST_PUBLIC_KEY) + processed_location = manager_function.process_location( + b''.join((selector, validator)), TEST_PUBLIC_KEY + ) assert location is processed_location assert isinstance(processed_location.b64encoded_pubkey, str) assert processed_location.b64encoded_pubkey == TEST_PUBLIC_KEY diff --git a/tox.ini b/tox.ini index cc4a542..6a686a1 100644 --- a/tox.ini +++ b/tox.ini @@ -90,7 +90,7 @@ exclude = .eggs format = ${cyan}%(path)s${reset}:${yellow_bold}%(row)d${reset}:${green_bold}%(col)d${reset}: ${red_bold}%(code)s${reset} %(text)s import-order-style = pycharm -ignore = D105,E501 +ignore = D105,E501,W503 max-complexity = 10 [testenv:doc8]