@@ -613,3 +613,224 @@ def from_csv(session, text: str, actor: str = "") -> dict:
     session.flush()
     log.info("tagging import: %s", counts)
     return {**counts, "problems": problems}
+
+
+# The columns the web applications import reads.  The rest of what the export
+# writes (visibility, last change, last scan and so on) is what a scan found,
+# so it is ignored on the way back in.
+WEB_IMPORT_FIELDS = ("application_name", "application_owner",
+                     "technical_owner", "sensitivity")
+
+# Small enough to ride along in the confirm page's form, which is how the
+# apply step gets exactly the file that was shown.
+MAX_WEB_IMPORT_BYTES = 512 * 1024
+
+
+def _web_cell(field: str, value: str) -> str:
+    """A cell the way the repository stores it, so a match reads as a match."""
+    if field in OWNERS:
+        return _owners(value)
+    if field == "sensitivity":
+        return ", ".join(split_labels(value))
+    return value
+
+
+def web_apps_plan(session, text: str, base=None) -> dict:
+    """What a web applications CSV would change.  Nothing is written.
+
+    The file is the whole list.  A repository in it is marked as part of a
+    web application with the addresses it gives, and one marked now that is
+    not in it is taken off, addresses and all.  Only repositories `base`
+    lets this person see are touched either way.
+
+    The export writes what applies, which may be the connection's answer.
+    So a cell that matches what applies is left alone, and only one that
+    differs becomes the repository's own value.  Otherwise every export and
+    import would copy the connection's owner onto each repository, and
+    nobody could tell later which ones had been meant.
+
+    -> {"rows", "adds", "updates", "removes", "same", "unknown",
+        "problems", "refused"}.  Each change is {"id", "repository",
+    "set": {field: (old, new)}}.
+    """
+    import csv
+    import io
+
+    out = {"rows": 0, "adds": [], "updates": [], "removes": [], "same": 0,
+           "unknown": [], "problems": [], "refused": ""}
+
+    try:
+        reader = csv.DictReader(io.StringIO(text))
+        header = [(name or "").strip().lower()
+                  for name in (reader.fieldnames or [])]
+    except (csv.Error, ValueError) as exc:
+        out["refused"] = f"could not read the file: {exc}"
+        return out
+    if "repository" not in header or "web_urls" not in header:
+        out["refused"] = ("that file needs repository and web_urls columns; "
+                          "export the web applications first to see the shape")
+        return out
+
+    query = base if base is not None else session.query(Repository)
+    visible = {row.full_name: row for row in query.all()}
+    applies = effective_many(session, list(visible.values()))
+    offered = labels(session)
+    for repo in visible.values():
+        offered += [x for x in split_labels(repo.sensitivity or "")
+                    if x not in offered]
+    by_case = {label.casefold(): label for label in offered}
+
+    listed = set()
+    try:
+        for raw in reader:
+            out["rows"] += 1
+            if out["rows"] > MAX_IMPORT_ROWS:
+                out["refused"] = f"more than {MAX_IMPORT_ROWS} rows"
+                return out
+            row = {(k or "").strip().lower():
+                   _unquote((str(v) if v is not None else "").strip())
+                   for k, v in raw.items() if k}
+            name = row.get("repository", "")
+            if not name:
+                continue
+            repo = visible.get(name)
+            if repo is None:
+                if name not in out["unknown"]:
+                    out["unknown"].append(name)
+                continue
+            if name in listed:
+                out["problems"].append(f"{name} is in the file twice; the "
+                                       "second row was left out")
+                continue
+            # Listed, even when the row turns out bad below.  A typo in one
+            # row must not read as "take this application off".
+            listed.add(name)
+
+            why, change = _web_row(repo, row, header,
+                                   applies.get(repo.id, {}), by_case, offered)
+            if why:
+                out["problems"].append(f"{name}: {why}")
+                continue
+            if not repo.web_app:
+                out["adds"].append(change)
+            elif change["set"]:
+                out["updates"].append(change)
+            else:
+                out["same"] += 1
+    except csv.Error as exc:
+        out["refused"] = f"could not read the file: {exc}"
+        return out
+
+    for name, repo in sorted(visible.items()):
+        if repo.web_app and name not in listed:
+            out["removes"].append({
+                "id": repo.id, "repository": name,
+                "set": {"web_app": (True, False),
+                        "web_app_urls": (repo.web_app_urls or "", "")}})
+
+    # An empty or wrong file would otherwise take every web application off.
+    if out["removes"] and not listed:
+        out["refused"] = ("the file names no web application you can see, so "
+                          "importing it would take every one off; nothing "
+                          "was changed")
+    return out
+
+
+def _web_row(repo, row, header, said, by_case, offered) -> tuple:
+    """One row of the web applications CSV, checked.  -> (why, change)."""
+    change = {"id": repo.id, "repository": repo.full_name, "set": {}}
+
+    given = row.get("web_urls", "")
+    urls, why = web_urls(given.replace("|", ","))
+    if why:
+        return f"web urls: {why}", None
+    if not urls:
+        return "a web application needs its web urls", None
+    new = ", ".join(urls)
+    if len(new) > LIMITS["web_app_urls"]:
+        return f"web urls is longer than {LIMITS['web_app_urls']}", None
+    if not repo.web_app:
+        change["set"]["web_app"] = (False, True)
+    if web_urls(repo.web_app_urls)[0] != urls:
+        change["set"]["web_app_urls"] = (repo.web_app_urls or "", new)
+
+    after = {}
+    for field in WEB_IMPORT_FIELDS:
+        if field not in header:
+            continue
+        given = row.get(field, "")
+        if given == "-":
+            given = ""
+        if field in OWNERS:
+            bad = bad_address(given.replace("|", ","))
+            if bad:
+                return (f"{field.replace('_', ' ')} has to be an email "
+                        f"address, and {bad!r} is not one"), None
+        elif field == "sensitivity":
+            held = []
+            for one in split_labels(given):
+                label = by_case.get(one.casefold())
+                if label is None:
+                    return (f"{one!r} is not one of the labels on the "
+                            f"Settings page ({', '.join(offered)})"), None
+                if label not in held:
+                    held.append(label)
+            given = ", ".join(held)
+        given = _web_cell(field, given)
+        if len(given) > LIMITS[field]:
+            return (f"{field.replace('_', ' ')} is longer than "
+                    f"{LIMITS[field]}"), None
+        applied = _web_cell(field, (said.get(field) or {}).get("value", ""))
+        own = (getattr(repo, field, "") or "").strip()
+        if given == applied:
+            continue
+        # An empty cell over an inherited answer, with nothing set here, is
+        # already as empty as this repository can make it.
+        if not given and not own:
+            continue
+        change["set"][field] = (own, given)
+        after[field] = given
+
+    # The same rule the repository's form holds to: a notify tick needs an
+    # owner to send to, its own or the connection's.
+    for flag, field in (("notify_application_owner", "application_owner"),
+                        ("notify_technical_owner", "technical_owner")):
+        if field not in after or after[field]:
+            continue
+        if (getattr(repo, flag, False)
+                and (said.get(field) or {}).get("from") != "connection"):
+            return (f"{flag.replace('_', ' ')} is on, but the file leaves no "
+                    f"{field.replace('_', ' ')} to send it to"), None
+    return "", change
+
+
+def web_apps_digest(plan: dict) -> str:
+    """A short fingerprint of a plan, so apply can tell it still holds."""
+    import hashlib
+    import json
+
+    body = json.dumps({side: [[c["id"], sorted(c["set"].items())]
+                              for c in plan[side]]
+                       for side in ("adds", "updates", "removes")},
+                      sort_keys=True, default=list)
+    return hashlib.sha256(body.encode("utf-8")).hexdigest()
+
+
+def web_apps_apply(session, plan: dict, actor: str = "") -> dict:
+    """Write what web_apps_plan worked out.  -> counts by side."""
+    ids = {c["id"]: c for side in ("adds", "updates", "removes")
+           for c in plan[side]}
+    rows = {row.id: row for row in session.query(Repository)
+            .filter(Repository.id.in_(list(ids))).all()} if ids else {}
+    for repo_id, change in ids.items():
+        repo = rows.get(repo_id)
+        if repo is None:
+            continue
+        for field, (_old, new) in change["set"].items():
+            setattr(repo, field, new)
+        repo.tagged_by = actor or repo.tagged_by or ""
+        repo.tagged_at = utcnow()
+    session.flush()
+    counts = {side: len(plan[side]) for side in ("adds", "updates", "removes")}
+    log.info("web applications import: %s", counts)
+    return counts
@@ -4170,6 +4170,55 @@ async def import_tagging(request: Request, user: User = Depends(current_user)):
                     + "&msg=" + quote(said[:290]))
 
 
+@router.post("/repositories/web-apps/import/apply")
+async def apply_web_apps_import(request: Request,
+                                user: User = Depends(current_user)):
+    """The web applications CSV, as the confirm page showed it.
+
+    The file is planned again from scratch.  If that plan is not the one the
+    page showed, because somebody changed a repository in the meantime,
+    nothing is written and the person is sent back to look again.
+    """
+    await check_csrf(request)
+    rbac.require(user, "repo.write")
+
+    from app.core import tagging
+
+    form = await request.form()
+    text = str(form.get("file_text") or "")
+    shown = str(form.get("digest") or "")
+    if not text or len(text.encode("utf-8")) > 2 * tagging.MAX_WEB_IMPORT_BYTES:
+        raise HTTPException(status_code=400, detail="Choose a file first.")
+
+    def work():
+        with session_scope() as db:
+            plan = tagging.web_apps_plan(
+                db, text, base=rbac.visible_repositories(db, user))
+            if plan["refused"]:
+                return None, plan["refused"]
+            if tagging.web_apps_digest(plan) != shown:
+                return None, ("Something changed since that page was shown.  "
+                              "Import the file again to see what it does now.")
+            counts = tagging.web_apps_apply(db, plan, actor=user.username)
+            names = lambda side: ", ".join(c["repository"]
+                                           for c in plan[side])
+            record(db, "webapps.imported", actor=user.username,
+                   detail=(f"{counts['adds']} added, {counts['updates']} "
+                           f"updated, {counts['removes']} removed.  "
+                           f"Added: {names('adds') or '-'}.  "
+                           f"Removed: {names('removes') or '-'}")[:4000],
+                   outcome="warn" if plan["problems"] else "ok",
+                   source_ip=client_ip(request))
+            return counts, ""
+
+    counts, why = await run_in_threadpool(work)
+    if counts is None:
+        return redirect(_back_with("/repositories", why, "warn"))
+    said = (f"Web applications imported: {counts['adds']} added, "
+            f"{counts['updates']} updated, {counts['removes']} removed.")
+    return redirect(_back_with("/repositories", said, "ok"))
+
+
 @router.post("/blast-radius/repo/{repo_id}/describe")
 async def describe_repository(repo_id: int, request: Request,
                               user: User = Depends(current_user)):
 
@@ -306,22 +306,33 @@ def export_csv(request: Request, user: User = Depends(current_user)):
         # One read of each site's newest counted scan, not one per column.
         scoped = external.scoping_by_asset(db)
         from app.core import dast_paths as _dast_paths
+        from app.core import webapps as _webapps
         for a in db.query(ExternalAsset).order_by(ExternalAsset.host):
             values = external.field_values(a)
             scope = (scoped.get(a.id) or {}).get("scope")
-            repos = []
+            repos, moved = [], []
             for l in mapper.links(a):
                 if l["linked"]:
                     repos += [f"{l['full_name']}@{b}" for b in l.get("branches") or []] \
                         or [l["full_name"]]
+                    days = _webapps.changed(l["repo"])["days"]
+                    if days is not None:
+                        moved.append(days)
+            # Typed on the site wins.  Otherwise the newest change on any
+            # linked repository's default branch, the same number the Web
+            # Applications tab shows on an application's line.
+            if a.change_frequency_days:
+                frequency = a.change_frequency_days
+            else:
+                frequency = min(moved) if moved else ""
             writer.writerow([_quote_formula(str(c)) for c in [
                 a.host, a.kind, a.parent_domain, a.application, "; ".join(a.tag_list()),
                 envs.get(a.environment, a.environment or ""),
                 external.STATUSES.get(a.status, a.status or ""),
                 a.executive_owner or "", a.technical_owner or "",
                 "yes" if a.external_auth_needed else "no",
                 "yes" if a.whitelist_needed else "no",
-                a.change_frequency_days or "",
+                frequency,
                 "yes" if a.no_scan else "", a.no_scan_reason or "",
                 a.telemetry_at.strftime("%Y-%m-%d %H:%M") if a.telemetry_at else "",
                 a.last_dast_at.strftime("%Y-%m-%d") if a.last_dast_at else "",
 
@@ -2344,6 +2344,71 @@ def export_web_apps(request: Request, user: User = Depends(current_user)):
                  f'attachment; filename="web-applications-{stamp}.csv"'})
 
 
+def _web_import_text(raw: bytes) -> str:
+    """An uploaded file as text, however the spreadsheet saved it."""
+    try:
+        return raw.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        return raw.decode("latin-1", "replace")
+
+
+def _web_import_plan(user, text: str):
+    """The plan for a web applications file, worked out off the event loop."""
+    from app.core import tagging
+
+    with session_scope() as db:
+        plan = tagging.web_apps_plan(db, text,
+                                     base=rbac.visible_repositories(db, user))
+        return plan, tagging.web_apps_digest(plan)
+
+
+@router.get("/repositories/web-apps/import")
+def web_apps_import_page(request: Request, user: User = Depends(current_user)):
+    """Where the web applications CSV goes back in."""
+    rbac.require(user, "repo.write")
+    with session_scope() as db:
+        context = _shared(db, user)
+        context.update(active_page="repositories")
+        return page(request, "web_apps_import.html", **context)
+
+
+@router.post("/repositories/web-apps/import")
+async def web_apps_import_check(request: Request,
+                                user: User = Depends(current_user)):
+    """What a web applications CSV would add, change and take off.
+
+    Nothing is written here.  The confirm page carries the file itself and a
+    fingerprint of what it showed, so apply can refuse if anything moved in
+    between.
+    """
+    from starlette.concurrency import run_in_threadpool
+
+    from app.core import tagging
+    from app.web.security import check_csrf
+
+    form = await request.form()
+    await check_csrf(request)
+    rbac.require(user, "repo.write")
+    upload = form.get("file")
+    if upload is None or not hasattr(upload, "read"):
+        raise HTTPException(status_code=400, detail="Choose a file first.")
+    raw = await upload.read()
+    if len(raw) > tagging.MAX_WEB_IMPORT_BYTES:
+        raise HTTPException(
+            status_code=413,
+            detail=f"That file is larger than "
+                   f"{tagging.MAX_WEB_IMPORT_BYTES // 1024}KB.")
+    text = _web_import_text(raw)
+    plan, digest = await run_in_threadpool(_web_import_plan, user, text)
+    with session_scope() as db:
+        context = _shared(db, user)
+        context.update(active_page="repositories", plan=plan, digest=digest,
+                       file_text=text,
+                       changes=(len(plan["adds"]) + len(plan["updates"])
+                                + len(plan["removes"])))
+        return page(request, "web_apps_import_confirm.html", **context)
+
+
 @router.get("/repositories/{repo_id:int}")
 def repository_about(repo_id: int, request: Request, msg: str = "",
                      kind: str = "", user: User = Depends(current_user)):
 
@@ -18,6 +18,10 @@
   {% endif %}
   <a class="btn" href="{{ base }}/repositories/web-apps.csv"
      title="Every repository marked as part of a web application, with its addresses and owners">Export web applications</a>
+  {% if can_tag %}
+  <a class="btn" href="{{ base }}/repositories/web-apps/import"
+     title="Bring the web applications file back.  Rows update what is there, and web applications not in the file are taken off">Import web applications</a>
+  {% endif %}
   {% if certified and not certified_filter %}
   <a class="btn" href="#certifications">Certified branches ({{ certified | length }})</a>
   {% endif %}</div></div>