Merge pull request #91 from internetarchive/fix-semgrep

Fix identifiers with page suffixes not working

iiify/app.py

@@ -81,7 +81,7 @@ def documentation():
 
 @app.route('/iiif/helper/<identifier>/')
 def helper(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=false)
 
     metadata = requests.get('%s/metadata/%s' % (ARCHIVE, identifier)).json()
 
@@ -111,7 +111,7 @@ def helper(identifier):
 
 @app.route('/iiif/<identifier>')
 def view(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
 
     domain = purify_domain(request.args.get('domain', request.url_root))
     uri = '%s%s' % (domain, identifier)
@@ -131,7 +131,7 @@ def view(identifier):
 @app.route('/iiif/3/<identifier>/collection.json')
 @cache.cached(timeout=cache_timeouts["med"], forced_update=cache_bust)
 def collection3JSON(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     domain = purify_domain(request.args.get('domain', request.url_root))
 
     try:
@@ -149,7 +149,7 @@ def collection3JSON(identifier):
 @app.route('/iiif/3/<identifier>/<page>/collection.json')
 @cache.cached(timeout=cache_timeouts["med"], forced_update=cache_bust)
 def collection3page(identifier, page):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     domain = purify_domain(request.args.get('domain', request.url_root))
 
     try:
@@ -168,21 +168,21 @@ def collection3page(identifier, page):
 @app.route('/iiif/<identifier>/collection.json')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def collectionJSON(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     return redirect(f'/iiif/3/{identifier}/collection.json', code=302)
 
 
 @app.route('/iiif/<identifier>/<page>/collection.json')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def collectionPage(identifier, page):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     return redirect(f'/iiif/3/{identifier}/{page}/collection.json', code=302)
 
 
 @app.route('/iiif/3/<identifier>/manifest.json')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def manifest3(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
 
     domain = purify_domain(request.args.get('domain', request.url_root))
     page = None
@@ -201,27 +201,27 @@ def manifest3(identifier):
 @app.route('/iiif/<int:version>/annotations/<identifier>/<fileName>/<int:canvas_no>.json')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def annnotations(version: str, identifier: str, fileName: str, canvas_no: int):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     domain = purify_domain(request.args.get('domain', request.url_root))
     return ldjsonify(create_annotations(version, identifier, fileName, canvas_no, domain=domain))
 
 @app.route('/iiif/vtt/streaming/<identifier>.vtt')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def vtt_stream(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=False)
     response = make_response(create_vtt_stream(identifier))
     response.headers['Content-Type'] = 'text/vtt'
     return response
 
 @app.route('/iiif/<identifier>/manifest.json')
 @cache.cached(timeout=cache_timeouts["long"], forced_update=cache_bust)
 def manifest(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     return redirect(f'/iiif/3/{identifier}/manifest.json', code=302)
 
 @app.route('/iiif/2/<identifier>/manifest.json')
 def manifest2(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     domain = purify_domain(request.args.get('domain', request.url_root))
     page = None
     if '$' in identifier:
@@ -237,28 +237,28 @@ def manifest2(identifier):
 
 @app.route('/iiif/<identifier>/info.json')
 def info(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     cantaloupe_id = cantaloupe_resolver(identifier)
     cantaloupe_url = f"{image_server}/2/{cantaloupe_id}/info.json"
     return redirect(cantaloupe_url, code=302)
 
 @app.route('/iiif/3/<identifier>/info.json')
 def info3(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     cantaloupe_id = cantaloupe_resolver(identifier)
     cantaloupe_url = f"{image_server}/3/{cantaloupe_id}/info.json"
     return redirect(cantaloupe_url, code=302)
 
 @app.route('/iiif/2/<identifier>/info.json')
 def info2(identifier):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     cantaloupe_id = cantaloupe_resolver(identifier)
     cantaloupe_url = f"{image_server}/2/{cantaloupe_id}/info.json"
     return redirect(cantaloupe_url, code=302)
 
 @app.route('/iiif/<identifier>/<region>/<size>/<rotation>/<quality>.<fmt>')
 def image_processor(identifier, region, size, rotation, quality, fmt):
-    validate_ia_identifier(identifier)
+    validate_ia_identifier(identifier, page_suffix=True)
     cantaloupe_id = cantaloupe_resolver(identifier)
     cantaloupe_url = f"{image_server}/2/{cantaloupe_id}/{region}/{size}/{rotation}/{quality}.{fmt}"
     return redirect(cantaloupe_url, code=302)
@@ -275,9 +275,15 @@ def ldjsonify(data):
     j.mimetype = "application/ld+json"
     return j
 
-def validate_ia_identifier(identifier: str) -> None:
+def validate_ia_identifier(identifier: str, page_suffix: bool) -> None:
+    if page_suffix:
+        if not re.match(r'^[a-zA-Z0-9_.-]{1,100}(\$\d+)?$', identifier):
+            abort(400, "Invalid identifier")
+            return
+
     if not re.match(r'^[a-zA-Z0-9_.-]{1,100}$', identifier):
         abort(400, "Invalid identifier")
+        return
 
 if __name__ == '__main__':
     app.run(**options)