#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2023 Robin Jarry
"""
Query a CardDAV server for contact names and emails.
"""
import argparse
import base64
import configparser
import os
import re
import subprocess
import sys
import xml.etree.ElementTree as xml
from urllib import error, parse, request
def main():
try:
args = parse_args()
C = "urn:ietf:params:xml:ns:carddav"
D = "DAV:"
xml.register_namespace("C", C)
xml.register_namespace("D", D)
# perform the actual address book query
query = xml.Element(f"{{{C}}}addressbook-query")
prop = xml.SubElement(query, f"{{{D}}}prop")
xml.SubElement(prop, f"{{{D}}}getetag")
data = xml.SubElement(prop, f"{{{C}}}address-data")
xml.SubElement(data, f"{{{C}}}prop", name="FN")
xml.SubElement(data, f"{{{C}}}prop", name="EMAIL")
limit = xml.SubElement(query, f"{{{C}}}limit")
xml.SubElement(limit, f"{{{C}}}nresults").text = str(args.limit)
filtre = xml.SubElement(query, f"{{{C}}}filter", test="anyof")
for term in args.terms:
for attr in "FN", "EMAIL", "NICKNAME", "ORG", "TITLE":
prop = xml.SubElement(filtre, f"{{{C}}}prop-filter", name=attr)
match = xml.SubElement(
prop, f"{{{C}}}text-match", {"match-type": "contains"}
)
match.text = term
data = http_request_xml(
"REPORT",
args.server_url,
query,
username=args.username,
password=args.password,
debug=args.verbose,
Depth="1",
)
for vcard in data.iterfind(f".//{{{C}}}address-data"):
for name, email in parse_vcard(vcard.text.strip()):
print(f"{email}\t{name}")
except Exception as e:
if isinstance(e, error.HTTPError):
if args.verbose:
debug_response(e.fp)
e = e.fp.read().decode()
print(f"error: {e}", file=sys.stderr)
sys.exit(1)
def http_request_xml(
method: str,
url: str,
data: xml.Element,
username: str = None,
password: str = None,
debug: bool = False,
**headers,
) -> xml.Element:
req = request.Request(
url=url,
method=method,
headers={
"Content-Type": 'text/xml; charset="utf-8"',
**headers,
},
data=xml.tostring(data, encoding="utf-8", xml_declaration=True),
)
if username is not None and password is not None:
auth = f"{username}:{password}"
auth = base64.standard_b64encode(auth.encode("utf-8")).decode("ascii")
req.add_header("Authorization", f"Basic {auth}")
if debug:
uri = parse.urlparse(req.full_url)
print(f"> {req.method} {uri.path} HTTP/1.1", file=sys.stderr)
print(f"> Host: {uri.hostname}", file=sys.stderr)
for name, value in req.headers.items():
print(f"> {name}: {value}", file=sys.stderr)
print(f"{req.data.decode('utf-8')}\n", file=sys.stderr)
with request.urlopen(req) as resp:
data = resp.read().decode("utf-8")
if debug:
debug_response(resp)
print(f"{data}", file=sys.stderr)
return xml.fromstring(data)
def debug_response(resp):
print(f"< HTTP/1.1 {resp.code}", file=sys.stderr)
for name, value in resp.headers.items():
print(f"< {name}: {value}", file=sys.stderr)
def parse_vcard(txt):
lines = txt.splitlines()
if len(lines) < 4 or lines[0] != "BEGIN:VCARD" or lines[-1] != "END:VCARD":
return
name = None
emails = []
for line in lines[1:-1]:
if line.startswith("FN:"):
name = line[len("FN:") :].replace("\\,", ",")
continue
match = re.match(r"^(?:ITEM\d+\.)?EMAIL(?:;[\w-]+=[^;:]+)*:(.+@.+)$", line)
if match:
email = match.group(1).lower().replace("\\,", ",")
if email not in emails:
if "TYPE=pref" in line or "PREF=1" in line:
emails.insert(0, email)
else:
emails.append(email)
if name is not None:
for e in emails:
yield name, e
def parse_args():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"-l",
"--limit",
default=10,
type=int,
help="""
Maximum number of results returned by the server (default: 10).
If the server does not support limiting, this will be disregarded.
""",
)
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="""
Print debug info on stderr.
""",
)
parser.add_argument(
"-c",
"--config-file",
metavar="FILE",
default=os.path.expanduser("~/.config/aerc/accounts.conf"),
help="""
INI configuration file from which to read the CardDAV URL endpoint
(default: ~/.config/aerc/accounts.conf).
""",
)
parser.add_argument(
"-S",
"--config-section",
metavar="SECTION",
help="""
INI configuration section where to find CONFIG_KEY. By default the
first section where CONFIG_KEY is found will be used.
""",
)
parser.add_argument(
"-k",
"--config-key-source",
metavar="KEY_SOURCE",
default="carddav-source",
help="""
INI configuration key to lookup in CONFIG_SECTION from CONFIG_FILE.
The value must respect the following format:
https?://USERNAME[:PASSWORD]@HOSTNAME/PATH/TO/ADDRESSBOOK.
Both USERNAME and PASSWORD must be percent encoded.
""",
)
parser.add_argument(
"-C",
"--config-key-cred-cmd",
metavar="KEY_CRED_CMD",
default="carddav-source-cred-cmd",
help="""
INI configuration key to lookup in CONFIG_SECTION from CONFIG_FILE. The
value is a command that will be used to determine PASSWORD if it is not
present in CONFIG_KEY_SOURCE.
""",
)
parser.add_argument(
"-s",
"--server-url",
help="""
CardDAV server URL endpoint. Overrides configuration file.
""",
)
parser.add_argument(
"-u",
"--username",
help="""
Username to authenticate on the server. Overrides configuration file.
""",
)
parser.add_argument(
"-p",
"--password",
help="""
Password for the specified user. Overrides configuration file.
""",
)
parser.add_argument(
"terms",
nargs="+",
metavar="TERM",
help="""
Search term. Will be used to search contacts from their FN (formatted
name), EMAIL, NICKNAME, ORG (company) and TITLE fields.
""",
)
args = parser.parse_args()
cfg = configparser.RawConfigParser(strict=False)
cfg.read([args.config_file])
source = cred_cmd = None
if args.config_section:
source = cfg.get(args.config_section, args.config_key_source, fallback=None)
cred_cmd = cfg.get(args.config_section, args.config_key_cred_cmd, fallback=None)
else:
for sec in cfg.sections():
source = cfg.get(sec, args.config_key_source, fallback=None)
if source is not None:
cred_cmd = cfg.get(sec, args.config_key_cred_cmd, fallback=None)
break
if source is not None:
try:
u = parse.urlparse(source)
if args.username is None:
args.username = u.username
if args.password is None:
args.password = u.password
if not args.password and cred_cmd is not None:
args.password = subprocess.check_output(
cred_cmd, shell=True, text=True, encoding="utf-8"
).strip()
if args.server_url is None:
args.server_url = f"{u.scheme}://{u.hostname}"
if u.port is not None:
args.server_url += f":{u.port}"
args.server_url += u.path
except ValueError as e:
parser.error(f"{args.config_file}: {e}")
if args.server_url is None:
parser.error("SERVER_URL is required")
return args
if __name__ == "__main__":
main()
